Elasticsearch核心API之reindex API的应用及原理分析

POST _reindex
{
  "source": {
    "index": "my-index-000001"
  },
  "dest": {
    "index": "my-new-index-000001"
  }
}

POST _reindex
{
  "source": {
    "index": "my-index-000001",
    "slice": {
      "id": 0,
      "max": 2
    }
  },
  "dest": {
    "index": "my-new-index-000001"
  }
}
POST _reindex
{
  "source": {
    "index": "my-index-000001",
    "slice": {
      "id": 1,
      "max": 2
    }
  },
  "dest": {
    "index": "my-new-index-000001"
  }
}

POST _reindex?slices=5&refresh
{
  "source": {
    "index": "my-index-000001"
  },
  "dest": {
    "index": "my-new-index-000001"
  }
}

POST _reindex
{
  "source": {
    "index": "source",
    "query": {
      "match": {
        "company": "cat"
      }
    }
  },
  "dest": {
    "index": "dest",
    "routing": "=cat"
  }
}

# 指定batch_size进行reindex的文档路由。
POST _reindex
{
  "source": {
    "index": "source",
    "size": 100
  },
  "dest": {
    "index": "dest",
    "routing": "=cat"
  }
}

POST _reindex
{
  "source": {
    "index": "source"
  },
  "dest": {
    "index": "dest",
    "pipeline": "some_ingest_pipeline"
  }
}

POST _reindex
{
  "source": {
    "index": "my-index-000001",
    "_source": ["user.id", "_doc"]
  },
  "dest": {
    "index": "my-new-index-000001"
  }
}

POST _reindex
{
  "source": {
    "index": "metricbeat-*"
  },
  "dest": {
    "index": "metricbeat"
  },
  "script": {
    "lang": "painless",
    "source": "ctx._index = 'metricbeat-' + (ctx._index.substring('metricbeat-'.length(), ctx._index.length())) + '-1'"
  }
}


# 使用script对文档进行修改。
POST _reindex
{
  "source": {
    "index": "my-index-000001"
  },
  "dest": {
    "index": "my-new-index-000001",
    "version_type": "external"
  },
  "script": {
    "source": "if (ctx._source.foo == 'bar') {ctx._version++; ctx._source.remove('foo')}",
    "lang": "painless"
  }
}

POST _reindex
{
  "source": {
    "remote": {
      "host": "http://otherhost:9200",
      "username": "user",
      "password": "pass"
    },
    "index": "my-index-000001",
    "query": {
      "match": {
        "test": "data"
      }
    }
  },
  "dest": {
    "index": "my-new-index-000001"
  }
}

public class ReindexAction extends ActionType<BulkByScrollResponse> {
    public static final ReindexAction INSTANCE = new ReindexAction();
    public static final String NAME = "indices:data/write/reindex";

    private ReindexAction() {
        super(NAME, BulkByScrollResponse::new);
    }
}

public ActionRequestValidationException validate() {
        ActionRequestValidationException e = super.validate();
        if (getSearchRequest().indices() == null || getSearchRequest().indices().length == 0) {
            e = addValidationError("use _all if you really want to copy from all existing indexes", e);
        }
        if (getSearchRequest().source().fetchSource() != null && getSearchRequest().source().fetchSource().fetchSource() == false) {
            e = addValidationError("_source:false is not supported in this context", e);
        }
        /*
         * Note that we don't call index's validator - it won't work because
         * we'll be filling in portions of it as we receive the docs. But we can
         * validate some things so we do that below.
         */
        if (destination.index() == null) {
            e = addValidationError("index must be specified", e);
            return e;
        }
        if (false == routingIsValid()) {
            e = addValidationError("routing must be unset, [keep], [discard] or [=<some new value>]", e);
        }
        if (destination.versionType() == INTERNAL) {
            if (destination.version() != Versions.MATCH_ANY && destination.version() != Versions.MATCH_DELETED) {
                e = addValidationError("unsupported version for internal versioning [" + destination.version() + ']', e);
            }
        }
        if (getRemoteInfo() != null) {
            if (getSearchRequest().source().query() != null) {
                e = addValidationError("reindex from remote sources should use RemoteInfo's query instead of source's query", e);
            }
            if (getSlices() == AbstractBulkByScrollRequest.AUTO_SLICES || getSlices() > 1) {
                e = addValidationError("reindex from remote sources doesn't support slices > 1 but was [" + getSlices() + "]", e);
            }
        }
        return e;
    }

public TaskInfo taskInfoGivenSubtaskInfo(String localNodeId, List<TaskInfo> sliceInfo) {
        if (isLeader() == false) {
            throw new IllegalStateException("This task is not set to be a leader of other slice subtasks");
        }

        List<BulkByScrollTask.StatusOrException> sliceStatuses = Arrays.asList(
            new BulkByScrollTask.StatusOrException[leaderState.getSlices()]
        );
        for (TaskInfo t : sliceInfo) {
            BulkByScrollTask.Status status = (BulkByScrollTask.Status) t.status();
            sliceStatuses.set(status.getSliceId(), new BulkByScrollTask.StatusOrException(status));
        }
        Status status = leaderState.getStatus(sliceStatuses);
        return taskInfo(localNodeId, getDescription(), status);
    }

public void setWorkerCount(int slices) {
        if (isLeader()) {
            throw new IllegalStateException("This task is already a leader for other slice subtasks");
        }
        if (isWorker()) {
            throw new IllegalStateException("This task is already a worker");
        }

        leaderState = new LeaderBulkByScrollTaskState(this, slices);
    }

public void setWorker(float requestsPerSecond, @Nullable Integer sliceId) {
        if (isWorker()) {
            throw new IllegalStateException("This task is already a worker");
        }
        if (isLeader()) {
            throw new IllegalStateException("This task is already a leader for other slice subtasks");
        }

        workerState = new WorkerBulkByScrollTaskState(this, sliceId, requestsPerSecond);
        if (isCancelled()) {
            workerState.handleCancel();
        }
    }

  public void delayPrepareBulkRequest(
        ThreadPool threadPool,
        long lastBatchStartTimeNS,
        int lastBatchSize,
        AbstractRunnable prepareBulkRequestRunnable
    ) {
        // Synchronize so we are less likely to schedule the same request twice.
        synchronized (delayedPrepareBulkRequestReference) {
            TimeValue delay = throttleWaitTime(lastBatchStartTimeNS, System.nanoTime(), lastBatchSize);
            logger.debug("[{}]: preparing bulk request for [{}]", task.getId(), delay);
            try {
                delayedPrepareBulkRequestReference.set(
                    new DelayedPrepareBulkRequest(threadPool, getRequestsPerSecond(), delay, new RunOnce(prepareBulkRequestRunnable))
                );
            } catch (EsRejectedExecutionException e) {
                prepareBulkRequestRunnable.onRejection(e);
            }
        }
    }

class DelayedPrepareBulkRequest {
        private final ThreadPool threadPool;
        private final Runnable command;
        private final float requestsPerSecond;
        private final Scheduler.ScheduledCancellable scheduled;

        DelayedPrepareBulkRequest(ThreadPool threadPool, float requestsPerSecond, TimeValue delay, Runnable command) {
            this.threadPool = threadPool;
            this.requestsPerSecond = requestsPerSecond;
            this.command = command;
            this.scheduled = threadPool.schedule(() -> {
                throttledNanos.addAndGet(delay.nanos());
                command.run();
            }, delay, ThreadPool.Names.GENERIC);
        }

        DelayedPrepareBulkRequest rethrottle(float newRequestsPerSecond) {
            if (newRequestsPerSecond < requestsPerSecond) {
                /* The user is attempting to slow the request down. We'll let the
                 * change in throttle take effect the next time we delay
                 * prepareBulkRequest. We can't just reschedule the request further
                 * out in the future because the bulk context might time out. */
                logger.debug(
                    "[{}]: skipping rescheduling because the new throttle [{}] is slower than the old one [{}]",
                    task.getId(),
                    newRequestsPerSecond,
                    requestsPerSecond
                );
                return this;
            }

            long remainingDelay = scheduled.getDelay(TimeUnit.NANOSECONDS);
            // Actually reschedule the task
            if (scheduled == null || false == scheduled.cancel()) {
                // Couldn't cancel, probably because the task has finished or been scheduled. Either way we have nothing to do here.
                logger.debug("[{}]: skipping rescheduling because we couldn't cancel the task", task.getId());
                return this;
            }

            /* Strangely enough getting here doesn't mean that you actually
             * cancelled the request, just that you probably did. If you stress
             * test it you'll find that requests sneak through. So each request
             * is given a runOnce boolean to prevent that. */
            TimeValue newDelay = newDelay(remainingDelay, newRequestsPerSecond);
            logger.debug("[{}]: rescheduling for [{}] in the future", task.getId(), newDelay);
            return new DelayedPrepareBulkRequest(threadPool, requestsPerSecond, newDelay, command);
        }

        /**
         * Scale back remaining delay to fit the new delay.
         */
        TimeValue newDelay(long remainingDelay, float newRequestsPerSecond) {
            if (remainingDelay < 0) {
                return timeValueNanos(0);
            }
            return timeValueNanos(round(remainingDelay * requestsPerSecond / newRequestsPerSecond));
        }
    }