[GitHub] [ozone] sodonnel commented on a diff in pull request #3337: HDDS-6598. Add a BackgroundPipelineScrubber to scrub all pipelines.

[GitBox <gi...@apache.org>]

sodonnel commented on code in PR #3337:
URL: https://github.com/apache/ozone/pull/3337#discussion_r857707839


##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineScrubber.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package org.apache.hadoop.hdds.scm.pipeline;
+
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.ha.SCMService;
+import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Background service to clean up pipelines with following conditions.
+ * - CLOSED
+ * - ALLOCATED for too long
+ */
+public class BackgroundPipelineScrubber implements SCMService {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(BackgroundPipelineScrubber.class);
+
+  private static final String THREAD_NAME = "PipelineScrubberThread";
+
+  private final PipelineManager pipelineManager;
+  private final ConfigurationSource conf;
+  private final SCMContext scmContext;
+
+  private final Lock serviceLock = new ReentrantLock();
+  private ServiceStatus serviceStatus = ServiceStatus.PAUSING;
+
+  private final AtomicBoolean running = new AtomicBoolean(false);
+  private Thread scrubThread;
+  private final long intervalInMillis;
+  private final long waitTimeInMillis;
+  private long lastTimeToBeReadyInMillis = 0;
+
+  public BackgroundPipelineScrubber(PipelineManager pipelineManager,
+      ConfigurationSource conf, SCMServiceManager serviceManager,
+      SCMContext scmContext) {
+    this.pipelineManager = pipelineManager;
+    this.conf = conf;
+    this.scmContext = scmContext;
+
+    this.intervalInMillis = conf.getTimeDuration(
+        ScmConfigKeys.OZONE_SCM_PIPELINE_SCRUB_INTERVAL,
+        ScmConfigKeys.OZONE_SCM_PIPELINE_SCRUB_INTERVAL_DEFAULT,
+        TimeUnit.MILLISECONDS);
+    this.waitTimeInMillis = conf.getTimeDuration(
+        HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT,
+        HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT_DEFAULT,
+        TimeUnit.MILLISECONDS);
+
+    serviceManager.register(this);
+
+    start();
+  }
+
+  @Override
+  public void notifyStatusChanged() {
+    serviceLock.lock();
+    try {
+      if (scmContext.isLeaderReady() && !scmContext.isInSafeMode()) {
+        if (serviceStatus != ServiceStatus.RUNNING) {
+          LOG.info("Service {} transitions to RUNNING.", getServiceName());
+          serviceStatus = ServiceStatus.RUNNING;
+          lastTimeToBeReadyInMillis = Time.monotonicNow();
+        }
+      } else {
+        if (serviceStatus != ServiceStatus.PAUSING) {
+          LOG.info("Service {} transitions to PAUSING.", getServiceName());
+          serviceStatus = ServiceStatus.PAUSING;
+        }
+      }
+    } finally {
+      serviceLock.unlock();
+    }
+  }
+
+  @Override
+  public boolean shouldRun() {
+    serviceLock.lock();
+    try {
+      // If safe mode is off, then this SCMService starts to run with a delay.
+      return serviceStatus == ServiceStatus.RUNNING &&
+          Time.monotonicNow() - lastTimeToBeReadyInMillis >= waitTimeInMillis;
+    } finally {
+      serviceLock.unlock();
+    }
+  }
+
+  @Override
+  public String getServiceName() {
+    return BackgroundPipelineScrubber.class.getSimpleName();
+  }
+
+  @Override
+  public void start() {
+    if (!running.compareAndSet(false, true)) {
+      LOG.info("Pipeline Scrubber Service is already running, skip start.");
+      return;
+    }
+    LOG.info("Starting Pipeline Scrubber Service.");
+
+    scrubThread = new Thread(this::run);
+    scrubThread.setName(THREAD_NAME);
+    scrubThread.setDaemon(true);
+    scrubThread.start();
+  }
+
+  @Override
+  public void stop() {
+    synchronized (this) {
+      if (!running.compareAndSet(true, false)) {
+        LOG.info("Pipeline Scrubber Service is not running, skip stop.");
+        return;
+      }
+      notifyAll();
+    }
+    LOG.info("Stopping Pipeline Scrubber Service.");
+
+    try {
+      scrubThread.join();
+      scrubThread = null;
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private void run() {
+    while (running.get()) {
+      try {
+        if (shouldRun()) {
+          scrubAllPipelines();
+        }
+        synchronized (this) {
+          wait(intervalInMillis);
+        }
+      } catch (InterruptedException e) {
+        LOG.warn("{} is interrupted.", THREAD_NAME);
+        Thread.currentThread().interrupt();
+      }
+    }
+  }
+
+  private void scrubAllPipelines() {

Review Comment:
   Here I wonder if we should change the PipelineManager.scrubPipelines method to just scrub all pipelines, rather than doing it based on RepConfig?
   
   In PipelineStateMap, to get all pipelines is easier than getting them repConfig by repConfig, as getting a set of repConfig, it needs to scan over all of them and filter out the ones to keep.
   
   ```
     /**
      * Get list of pipelines in SCM.
      * @return List of pipelines
      */
     public List<Pipeline> getPipelines() {
       return new ArrayList<>(pipelineMap.values());
     }
   
     /**
      * Get pipeline corresponding to specified replication type.
      *
      * @param replicationConfig - ReplicationConfig
      * @return List of pipelines which have the specified replication type
      */
     List<Pipeline> getPipelines(ReplicationConfig replicationConfig) {
       Preconditions
           .checkNotNull(replicationConfig, "ReplicationConfig cannot be null");
   
       List<Pipeline> pipelines = new ArrayList<>();
       for (Pipeline pipeline : pipelineMap.values()) {
         if (pipeline.getReplicationConfig().equals(replicationConfig)) {
           pipelines.add(pipeline);
         }
       }
   
       return pipelines;
     }
   ```
   
   I don't see any good reason to scrub pipelines by repConfig - it would be simpler and slightly more efficient to just scrub all pipelines - what do you think?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org