【源码解析】Solr Overseer角色

古城微笑少年丶 2022-04-11 11:41 113阅读 0赞

NOTE: Mostly based on [Solr 7.2][]  
首先来看下Solr官方对Overseer类的注解：

/**
     * Cluster leader. Responsible for processing state updates, node assignments, creating/deleting
     * collections, shards, replicas and setting various properties.
     */

> 字面上的意思是，Overseer是集群的领导者，负责处理集群状态变更，节点任务，创建删除索引、分片和副本，并支持设置多种属性。Overseer角色对于Solr集群的重要性不言而喻。

// overseer not responsible for closing reader
      public Overseer(ShardHandler shardHandler, UpdateShardHandler updateShardHandler, String adminPath,
          final ZkStateReader reader, ZkController zkController, CloudConfig config) throws KeeperException, InterruptedException {
        this.reader = reader;
        this.shardHandler = shardHandler;
        this.updateShardHandler = updateShardHandler;
        this.adminPath = adminPath;
        this.zkController = zkController;
        this.stats = new Stats();
        this.config = config;
      }

这是Overseer唯一的构造函数，在ZKController类中被唯一调用。ZKController的定位是用来Handle ZooKeeper interactions。  
Overseer中定义了一个子类ClusterStateUpdater，顾名思义，是用来更新集群状态的。子类ClusterStateUpdater共计约400行代码，几乎占用整个Overseer类的一半篇幅，从代码量上可以想见这个类的重要性。  
来，贴上完整的代码看看。*重要的地方以 “ // —— 中文注解 ” 的形式标出。*

private class ClusterStateUpdater implements Runnable, Closeable {
        
        private final ZkStateReader reader;
        private final SolrZkClient zkClient;
        private final String myId;
        //queue where everybody can throw tasks
        private final ZkDistributedQueue stateUpdateQueue; // —— 任何人都可以往里面丢任务的通用队列，包括状态变更，创建索引等任务。
        //TODO remove in 9.0, we do not push message into this queue anymore
        //Internal queue where overseer stores events that have not yet been published into cloudstate
        //If Overseer dies while extracting the main queue a new overseer will start from this queue 
        private final ZkDistributedQueue workQueue; // —— 这个workQueue在Solr 9.0将会被移除，这里的作用仅仅是replay队列/overseer/queue-work里面的任务one by one。
        // Internal map which holds the information about running tasks.
        private final DistributedMap runningMap;
        // Internal map which holds the information about successfully completed tasks.
        private final DistributedMap completedMap;
        // Internal map which holds the information about failed tasks.
        private final DistributedMap failureMap;
    
        private final Stats zkStats;
    
        private boolean isClosed = false;
    
        // —— ClusterStateUpdater构造函数
        public ClusterStateUpdater(final ZkStateReader reader, final String myId, Stats zkStats) {
          this.zkClient = reader.getZkClient();
          this.zkStats = zkStats;
          this.stateUpdateQueue = getStateUpdateQueue(zkClient, zkStats);
          this.workQueue = getInternalWorkQueue(zkClient, zkStats);
          this.failureMap = getFailureMap(zkClient);
          this.runningMap = getRunningMap(zkClient);
          this.completedMap = getCompletedMap(zkClient);
          this.myId = myId;
          this.reader = reader;
        }
    
        public Stats getStateUpdateQueueStats() {
          return stateUpdateQueue.getZkStats();
        }
    
        public Stats getWorkQueueStats()  {
          return workQueue.getZkStats();
        }
        
        @Override
        public void run() { // —— 一大波僵尸正在靠近~BOOM~~~
    
          LeaderStatus isLeader = amILeader();
          while (isLeader == LeaderStatus.DONT_KNOW) {
            log.debug("am_i_leader unclear {}", isLeader);
            isLeader = amILeader();  // not a no, not a yes, try ask again —— 这个注解已经很明显了，跳出循环的条件就是根据Overseer实例id去判断是leader或者不是leader，而不再是DONT_KNOW状态。
          }
    
          log.debug("Starting to work on the main queue");
          try {
            ZkStateWriter zkStateWriter = null;
            ClusterState clusterState = null;
            boolean refreshClusterState = true; // let's refresh in the first iteration —— 只在第一次迭代去处理
            // we write updates in batch, but if an exception is thrown when writing new clusterstate,
            // we do not sure which message is bad message, therefore we will re-process node one by one
            int fallbackQueueSize = Integer.MAX_VALUE;
            ZkDistributedQueue fallbackQueue = workQueue; // —— 注意fallbackQueue就是workQueue，即/solr/overseer/queue-work队列。
            while (!this.isClosed) {
              isLeader = amILeader();
              if (LeaderStatus.NO == isLeader) {
                break;
              }
              else if (LeaderStatus.YES != isLeader) {
                log.debug("am_i_leader unclear {}", isLeader);
                continue; // not a no, not a yes, try ask again —— 还是一样，这个注解已经很明显了，if的意思已经确定了Overseer状态是NO的直接跳出循环；else if是说如果不是YES，则再次循环，其实就只有YES状态的Overseer才能继续往下走。
              }
    
              //TODO consider removing 'refreshClusterState' and simply check if clusterState is null
              if (refreshClusterState) {
                try {
                  reader.forciblyRefreshAllClusterStateSlow();
                  clusterState = reader.getClusterState();
                  zkStateWriter = new ZkStateWriter(reader, stats);
                  refreshClusterState = false; // —— 只在第一次迭代去处理，其后/solr/overseer/queue-work队列已处理完毕清空。
    
                  // if there were any errors while processing
                  // the state queue, items would have been left in the
                  // work queue so let's process those first
                  byte[] data = fallbackQueue.peek(); // —— one by one，一个个来处理！
                  while (fallbackQueueSize > 0 && data != null)  {
                    final ZkNodeProps message = ZkNodeProps.load(data);
                    log.debug("processMessage: fallbackQueueSize: {}, message = {}", fallbackQueue.getZkStats().getQueueLength(), message);
                    // force flush to ZK after each message because there is no fallback if workQueue items
                    // are removed from workQueue but fail to be written to ZK
                    try {
                      clusterState = processQueueItem(message, clusterState, zkStateWriter, false, null); // —— 里面没有采用batch，直接来一个处理一个，因为一般workqueue里面就一个。
                    } catch (Exception e) {
                      if (isBadMessage(e)) {
                        log.warn("Exception when process message = {}, consider as bad message and poll out from the queue", message);
                        fallbackQueue.poll();
                      }
                      throw e;
                    }
                    fallbackQueue.poll(); // poll-ing removes the element we got by peek-ing
                    data = fallbackQueue.peek();
                    fallbackQueueSize--;
                  }
                  // force flush at the end of the loop, if there are no pending updates, this is a no op call
                  clusterState = zkStateWriter.writePendingUpdates(); // —— 最后刷写一次到ZK
                  // the workQueue is empty now, use stateUpdateQueue as fallback queue
                  fallbackQueue = stateUpdateQueue;
                  fallbackQueueSize = 0;
                } catch (KeeperException.SessionExpiredException e) {
                  log.warn("Solr cannot talk to ZK, exiting Overseer work queue loop", e);
                  return;
                } catch (InterruptedException e) {
                  Thread.currentThread().interrupt();
                  return;
                } catch (Exception e) {
                  log.error("Exception in Overseer when process message from work queue, retrying", e);
                  refreshClusterState = true;
                  continue;
                }
              }
    
              // —— 接下来就是对/solr/overseer/queue的高效批量处理！！！
              LinkedList<Pair<String, byte[]>> queue = null;
              try {
                // We do not need to filter any nodes here cause all processed nodes are removed once we flush clusterstate
                queue = new LinkedList<>(stateUpdateQueue.peekElements(1000, 3000L, (x) -> true));  // —— 批量读取1000个消息
              } catch (KeeperException.SessionExpiredException e) {
                log.warn("Solr cannot talk to ZK, exiting Overseer main queue loop", e);
                return;
              } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
              } catch (Exception e) {
                log.error("Exception in Overseer main queue loop", e);
              }
              try {
                Set<String> processedNodes = new HashSet<>();
                while (queue != null && !queue.isEmpty()) {  // —— 逐个消息处理
                  for (Pair<String, byte[]> head : queue) {
                    byte[] data = head.second();
                    final ZkNodeProps message = ZkNodeProps.load(data);
                    log.debug("processMessage: queueSize: {}, message = {} current state version: {}", stateUpdateQueue.getZkStats().getQueueLength(), message, clusterState.getZkClusterStateVersion());
    
                    processedNodes.add(head.first());
                    fallbackQueueSize = processedNodes.size();
                    // The callback always be called on this thread
                    clusterState = processQueueItem(message, clusterState, zkStateWriter, true, () -> {  // —— true开启batch，压入队列等待批量往ZK写入。
                      stateUpdateQueue.remove(processedNodes);
                      processedNodes.clear();
                    });
                  }
                  if (isClosed) break;
                  // if an event comes in the next 100ms batch it together
                  queue = new LinkedList<>(stateUpdateQueue.peekElements(1000, 100, node -> !processedNodes.contains(node)));
                }
                fallbackQueueSize = processedNodes.size();
                // we should force write all pending updates because the next iteration might sleep until there
                // are more items in the main queue
                clusterState = zkStateWriter.writePendingUpdates();
                // clean work queue
                stateUpdateQueue.remove(processedNodes);
                processedNodes.clear();
              } catch (KeeperException.SessionExpiredException e) {
                log.warn("Solr cannot talk to ZK, exiting Overseer main queue loop", e);
                return;
              } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
              } catch (Exception e) {
                log.error("Exception in Overseer main queue loop", e);
                refreshClusterState = true; // it might have been a bad version error
              }
            }
          } finally {
            log.info("Overseer Loop exiting : {}", LeaderElector.getNodeName(myId));
            //do this in a separate thread because any wait is interrupted in this main thread
            new Thread(this::checkIfIamStillLeader, "OverseerExitThread").start();
          }
        }
    
        // Return true whenever the exception thrown by ZkStateWriter is correspond
        // to a invalid state or 'bad' message (in this case, we should remove that message from queue)
        private boolean isBadMessage(Exception e) {
          if (e instanceof KeeperException) {
            KeeperException ke = (KeeperException) e;
            return ke.code() == KeeperException.Code.NONODE || ke.code() == KeeperException.Code.NODEEXISTS;
          }
          return !(e instanceof InterruptedException);
        }
    
        private ClusterState processQueueItem(ZkNodeProps message, ClusterState clusterState, ZkStateWriter zkStateWriter, boolean enableBatching, ZkStateWriter.ZkWriteCallback callback) throws Exception {
          final String operation = message.getStr(QUEUE_OPERATION);
          List<ZkWriteCommand> zkWriteCommands = null;
          final Timer.Context timerContext = stats.time(operation);
          try {
            zkWriteCommands = processMessage(clusterState, message, operation);
            stats.success(operation);
          } catch (Exception e) {
            // generally there is nothing we can do - in most cases, we have
            // an issue that will fail again on retry or we cannot communicate with     a
            // ZooKeeper in which case another Overseer should take over
            // TODO: if ordering for the message is not important, we could
            // track retries and put it back on the end of the queue
            log.error("Overseer could not process the current clusterstate state update message, skipping the message: " + message, e);
            stats.error(operation);
          } finally {
            timerContext.stop();
          }
          if (zkWriteCommands != null) {
            clusterState = zkStateWriter.enqueueUpdate(clusterState, zkWriteCommands, callback);
            if (!enableBatching)  {
              clusterState = zkStateWriter.writePendingUpdates();
            }
          }
          return clusterState;
        }
    
        private void checkIfIamStillLeader() {
          ......
        }
    
        private List<ZkWriteCommand> processMessage(ClusterState clusterState,
            final ZkNodeProps message, final String operation) {
          CollectionParams.CollectionAction collectionAction = CollectionParams.CollectionAction.get(operation);
          if (collectionAction != null) {
            switch (collectionAction) {
              case CREATE:
                return Collections.singletonList(new ClusterStateMutator(getSolrCloudManager()).createCollection(clusterState, message));
              case DELETE:
                return Collections.singletonList(new ClusterStateMutator(getSolrCloudManager()).deleteCollection(clusterState, message));
              case CREATESHARD:
                return Collections.singletonList(new CollectionMutator(getSolrCloudManager()).createShard(clusterState, message));
              case DELETESHARD:
                return Collections.singletonList(new CollectionMutator(getSolrCloudManager()).deleteShard(clusterState, message));
              case ADDREPLICA:
                return Collections.singletonList(new SliceMutator(getSolrCloudManager()).addReplica(clusterState, message));
              case ADDREPLICAPROP:
                return Collections.singletonList(new ReplicaMutator(getSolrCloudManager()).addReplicaProperty(clusterState, message));
              case DELETEREPLICAPROP:
                return Collections.singletonList(new ReplicaMutator(getSolrCloudManager()).deleteReplicaProperty(clusterState, message));
              case BALANCESHARDUNIQUE:
                ExclusiveSliceProperty dProp = new ExclusiveSliceProperty(clusterState, message);
                if (dProp.balanceProperty()) {
                  String collName = message.getStr(ZkStateReader.COLLECTION_PROP);
                  return Collections.singletonList(new ZkWriteCommand(collName, dProp.getDocCollection()));
                }
                break;
              case MODIFYCOLLECTION:
                CollectionsHandler.verifyRuleParams(zkController.getCoreContainer() ,message.getProperties());
                return Collections.singletonList(new CollectionMutator(getSolrCloudManager()).modifyCollection(clusterState,message));
              case MIGRATESTATEFORMAT:
                return Collections.singletonList(new ClusterStateMutator(getSolrCloudManager()).migrateStateFormat(clusterState, message));
              default:
                throw new RuntimeException("unknown operation:" + operation
                    + " contents:" + message.getProperties());
            }
          } else {
            OverseerAction overseerAction = OverseerAction.get(operation);
            if (overseerAction == null) {
              throw new RuntimeException("unknown operation:" + operation + " contents:" + message.getProperties());
            }
            switch (overseerAction) {
              case STATE:
                return Collections.singletonList(new ReplicaMutator(getSolrCloudManager()).setState(clusterState, message));
              case LEADER:
                return Collections.singletonList(new SliceMutator(getSolrCloudManager()).setShardLeader(clusterState, message));
              case DELETECORE:
                return Collections.singletonList(new SliceMutator(getSolrCloudManager()).removeReplica(clusterState, message));
              case ADDROUTINGRULE:
                return Collections.singletonList(new SliceMutator(getSolrCloudManager()).addRoutingRule(clusterState, message));
              case REMOVEROUTINGRULE:
                return Collections.singletonList(new SliceMutator(getSolrCloudManager()).removeRoutingRule(clusterState, message));
              case UPDATESHARDSTATE:
                return Collections.singletonList(new SliceMutator(getSolrCloudManager()).updateShardState(clusterState, message));
              case QUIT:
                if (myId.equals(message.get(ID))) {
                  log.info("Quit command received {} {}", message, LeaderElector.getNodeName(myId));
                  overseerCollectionConfigSetProcessor.close();
                  close();
                } else {
                  log.warn("Overseer received wrong QUIT message {}", message);
                }
                break;
              case DOWNNODE:
                return new NodeMutator().downNode(clusterState, message);
              default:
                throw new RuntimeException("unknown operation:" + operation + " contents:" + message.getProperties());
            }
          }
    
          return Collections.singletonList(ZkStateWriter.NO_OP);
        }
    
        private LeaderStatus amILeader() {
          ......
        }
        ......
      }

[Solr 7.2]: http://archive.apache.org/dist/lucene/solr/7.2.0/changes/Changes.html