You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@druid.apache.org by "imply-cheddar (via GitHub)" <gi...@apache.org> on 2023/05/10 06:38:11 UTC

[GitHub] [druid] imply-cheddar opened a new pull request, #14239: Be able to load segments on Peons

imply-cheddar opened a new pull request, #14239:
URL: https://github.com/apache/druid/pull/14239

   This change introduces a new config on WorkerConfig that indicates how many bytes of each storage
   location to use for storage of a task.  Said config is divided up amongst the locations and slots
   and then used to set TaskConfig.tmpStorageBytesPerTask
   
   The Peons use their local task dir and
   tmpStorageBytesPerTask as their StorageLocations for the SegmentManager such that they can accept broadcast segments.
   
   TODO: this change requires docs updates before merge.  I'd like to get a sense for the tests and stuff first.
   
   #### Release note
   
   Peons now have the ability to store and work with segments broadcast to them.


-- 
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: commits-unsubscribe@druid.apache.org

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


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] abhishekagarwal87 commented on a diff in pull request #14239: Be able to load segments on Peons

Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.

abhishekagarwal87 commented on code in PR #14239:
URL: https://github.com/apache/druid/pull/14239#discussion_r1189674946


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/TaskStorageDirTracker.java:
##########
@@ -24,91 +24,202 @@
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
 
 import javax.annotation.Nullable;
 import java.io.File;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+/**
+ * Used to pick storage slots for tasks when run from the middle manager.
+ */
 public class TaskStorageDirTracker
 {
   public static TaskStorageDirTracker fromConfigs(WorkerConfig workerConfig, TaskConfig taskConfig)
   {
+    final List<File> baseTaskDirs;
     if (workerConfig == null) {
-      return new TaskStorageDirTracker(ImmutableList.of(taskConfig.getBaseTaskDir()));
+      baseTaskDirs = ImmutableList.of(taskConfig.getBaseTaskDir());
     } else {
       final List<String> basePaths = workerConfig.getBaseTaskDirs();
       if (basePaths == null) {
-        return new TaskStorageDirTracker(ImmutableList.of(taskConfig.getBaseTaskDir()));
+        baseTaskDirs = ImmutableList.of(taskConfig.getBaseTaskDir());
+      } else {
+        baseTaskDirs = basePaths.stream().map(File::new).collect(Collectors.toList());
       }
-      return new TaskStorageDirTracker(
-          basePaths.stream().map(File::new).collect(Collectors.toList())
-      );
     }
+
+    return fromBaseDirs(baseTaskDirs, workerConfig.getCapacity(), workerConfig.getBaseTaskDirSize());
+  }
+
+  public static TaskStorageDirTracker fromBaseDirs(List<File> baseTaskDirs, int numSlots, long dirSize)
+  {
+    int slotsPerBaseTaskDir = Math.max(1, numSlots / baseTaskDirs.size());
+    if (numSlots % baseTaskDirs.size() > 0) {
+      // We have to add an extra slot per location if they do not evenly divide
+      ++slotsPerBaseTaskDir;
+    }
+    long sizePerSlot = dirSize / slotsPerBaseTaskDir;
+
+    StorageSlot[] slots = new StorageSlot[numSlots];
+    for (int i = 0; i < numSlots; ++i) {
+      final int whichDir = i % baseTaskDirs.size();
+      final int dirUsageCount = i / baseTaskDirs.size();
+      final File slotDirectory = new File(baseTaskDirs.get(whichDir), StringUtils.format("slot%d", dirUsageCount));
+      slots[i] = new StorageSlot(slotDirectory, sizePerSlot);
+    }

Review Comment:
   It will be useful to clarify that we don't expect more slots than the number of base directories. For such a configuration, some base directories will remain unused since a task is allowed to use at most one directory. 



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/TaskStorageDirTracker.java:
##########
@@ -24,91 +24,202 @@
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
 
 import javax.annotation.Nullable;
 import java.io.File;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+/**
+ * Used to pick storage slots for tasks when run from the middle manager.
+ */
 public class TaskStorageDirTracker
 {
   public static TaskStorageDirTracker fromConfigs(WorkerConfig workerConfig, TaskConfig taskConfig)
   {
+    final List<File> baseTaskDirs;
     if (workerConfig == null) {
-      return new TaskStorageDirTracker(ImmutableList.of(taskConfig.getBaseTaskDir()));
+      baseTaskDirs = ImmutableList.of(taskConfig.getBaseTaskDir());
     } else {
       final List<String> basePaths = workerConfig.getBaseTaskDirs();
       if (basePaths == null) {
-        return new TaskStorageDirTracker(ImmutableList.of(taskConfig.getBaseTaskDir()));
+        baseTaskDirs = ImmutableList.of(taskConfig.getBaseTaskDir());
+      } else {
+        baseTaskDirs = basePaths.stream().map(File::new).collect(Collectors.toList());
       }
-      return new TaskStorageDirTracker(
-          basePaths.stream().map(File::new).collect(Collectors.toList())
-      );
     }
+
+    return fromBaseDirs(baseTaskDirs, workerConfig.getCapacity(), workerConfig.getBaseTaskDirSize());
+  }
+
+  public static TaskStorageDirTracker fromBaseDirs(List<File> baseTaskDirs, int numSlots, long dirSize)
+  {
+    int slotsPerBaseTaskDir = Math.max(1, numSlots / baseTaskDirs.size());
+    if (numSlots % baseTaskDirs.size() > 0) {
+      // We have to add an extra slot per location if they do not evenly divide
+      ++slotsPerBaseTaskDir;
+    }
+    long sizePerSlot = dirSize / slotsPerBaseTaskDir;

Review Comment:
   all directories might not be of same size though? it's ok we enforce/assume that. something to call out in the docs PR that the same amount space is reserved from all base directories.   



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/TaskStorageDirTracker.java:
##########
@@ -24,91 +24,202 @@
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
 
 import javax.annotation.Nullable;
 import java.io.File;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+/**
+ * Used to pick storage slots for tasks when run from the middle manager.
+ */
 public class TaskStorageDirTracker
 {
   public static TaskStorageDirTracker fromConfigs(WorkerConfig workerConfig, TaskConfig taskConfig)
   {
+    final List<File> baseTaskDirs;
     if (workerConfig == null) {
-      return new TaskStorageDirTracker(ImmutableList.of(taskConfig.getBaseTaskDir()));
+      baseTaskDirs = ImmutableList.of(taskConfig.getBaseTaskDir());
     } else {
       final List<String> basePaths = workerConfig.getBaseTaskDirs();
       if (basePaths == null) {
-        return new TaskStorageDirTracker(ImmutableList.of(taskConfig.getBaseTaskDir()));
+        baseTaskDirs = ImmutableList.of(taskConfig.getBaseTaskDir());
+      } else {
+        baseTaskDirs = basePaths.stream().map(File::new).collect(Collectors.toList());
       }
-      return new TaskStorageDirTracker(
-          basePaths.stream().map(File::new).collect(Collectors.toList())
-      );
     }
+
+    return fromBaseDirs(baseTaskDirs, workerConfig.getCapacity(), workerConfig.getBaseTaskDirSize());
+  }
+
+  public static TaskStorageDirTracker fromBaseDirs(List<File> baseTaskDirs, int numSlots, long dirSize)
+  {
+    int slotsPerBaseTaskDir = Math.max(1, numSlots / baseTaskDirs.size());
+    if (numSlots % baseTaskDirs.size() > 0) {
+      // We have to add an extra slot per location if they do not evenly divide
+      ++slotsPerBaseTaskDir;
+    }
+    long sizePerSlot = dirSize / slotsPerBaseTaskDir;
+
+    StorageSlot[] slots = new StorageSlot[numSlots];
+    for (int i = 0; i < numSlots; ++i) {
+      final int whichDir = i % baseTaskDirs.size();
+      final int dirUsageCount = i / baseTaskDirs.size();
+      final File slotDirectory = new File(baseTaskDirs.get(whichDir), StringUtils.format("slot%d", dirUsageCount));
+      slots[i] = new StorageSlot(slotDirectory, sizePerSlot);
+    }
+
+    return new TaskStorageDirTracker(baseTaskDirs, slots);
   }
 
+  /**
+   * The base task dirs, this field exists primarily for compatibility with scheduling that was done
+   * before TaskStorageDirTracker was introduced.  All of the tasks were just splatted together
+   * into one directory.  If we want to be able to restore the tasks, we need to be able to find them
+   * at the old locations and that is why this exists.
+   */
   private final File[] baseTaskDirs;
-  // Initialize to a negative number because it ensures that we can handle the overflow-rollover case
+
+  /**
+   * These are slots pre-divided to keep disk sizing considerations aligned.  The normal operation of this
+   * class is to round-robin across these slots.
+   */
+  private final StorageSlot[] slots;
+
+  /**
+   * A counter used to simplify round-robin allocation.  We initialize it to a negative value because it
+   * simplifies testing/ensuring that we can handle overflow-rollover of the integer
+   */
   private final AtomicInteger iterationCounter = new AtomicInteger(Integer.MIN_VALUE);
 
-  public TaskStorageDirTracker(List<File> baseTaskDirs)
+  public TaskStorageDirTracker(List<File> baseTaskDirs, StorageSlot[] slots)
   {
     this.baseTaskDirs = baseTaskDirs.toArray(new File[0]);
+    this.slots = slots;
   }
 
   @LifecycleStart
   public void ensureDirectories()
   {
-    for (File baseTaskDir : baseTaskDirs) {
-      if (!baseTaskDir.exists()) {
+    for (StorageSlot slot : slots) {
+      if (!slot.getDirectory().exists()) {
         try {
-          FileUtils.mkdirp(baseTaskDir);
+          FileUtils.mkdirp(slot.getDirectory());
         }
         catch (IOException e) {
           throw new ISE(
               e,
-              "base task directory [%s] likely does not exist, please ensure it exists and the user has permissions.",
-              baseTaskDir
+              "directory for slot [%s] likely does not exist, please ensure it exists and the user has permissions.",
+              slot
           );
         }
       }
     }
   }
 
-  public File pickBaseDir(String taskId)
+  public synchronized StorageSlot pickStorageSlot(String taskId)
   {
-    if (baseTaskDirs.length == 1) {
-      return baseTaskDirs[0];
+    // if the task directory already exists, we want to give it precedence, so check.
+    for (StorageSlot slot : slots) {
+      if (slot.runningTaskId != null && slot.runningTaskId.equals(taskId)) {
+        return slot;
+      }
     }
 
-    // if the task directory already exists, we want to give it precedence, so check.
-    for (File baseTaskDir : baseTaskDirs) {
-      if (new File(baseTaskDir, taskId).exists()) {
-        return baseTaskDir;
+    // if it doesn't exist, pick one round-robin and ensure it is unused.
+    for (int i = 0; i < slots.length; ++i) {
+      // This can be negative, so abs() it.
+      final int currIncrement = Math.abs(iterationCounter.getAndIncrement() % slots.length);
+      final StorageSlot candidateSlot = slots[currIncrement % slots.length];
+      if (candidateSlot.runningTaskId != null) {
+        continue;
       }
+      candidateSlot.runningTaskId = taskId;
+      return candidateSlot;
     }
+    throw new ISE("Unable to pick a free slot, this should never happen, slot status [%s].", Arrays.toString(slots));
+  }
 
-    // if it doesn't exist, pick one round-robin and return.  This can be negative, so abs() it
-    final int currIncrement = Math.abs(iterationCounter.getAndIncrement() % baseTaskDirs.length);
-    return baseTaskDirs[currIncrement % baseTaskDirs.length];
+  public synchronized void returnStorageSlot(StorageSlot slot)
+  {
+    slot.runningTaskId = null;
   }
 
   @Nullable
-  public File findExistingTaskDir(String taskId)
+  public synchronized File findExistingTaskDir(String taskId)
   {
+    File candidateLocation = null;
     if (baseTaskDirs.length == 1) {
-      return new File(baseTaskDirs[0], taskId);
+      candidateLocation = new File(baseTaskDirs[0], taskId);
+    } else {
+      for (File baseTaskDir : baseTaskDirs) {
+        File maybeExists = new File(baseTaskDir, taskId);
+        if (maybeExists.exists()) {
+          candidateLocation = maybeExists;
+          break;
+        }
+      }
     }
 
-    for (File baseTaskDir : baseTaskDirs) {
-      final File candidateLocation = new File(baseTaskDir, taskId);
-      if (candidateLocation.exists()) {
-        return candidateLocation;
+    if (candidateLocation != null && candidateLocation.exists()) {
+      // task exists at old location, relocate to a "good" slot location and return that.
+      final StorageSlot taskSlot = pickStorageSlot(taskId);
+      final File pickedLocation = new File(taskSlot.getDirectory(), taskId);
+      if (candidateLocation.renameTo(pickedLocation)) {
+        taskSlot.runningTaskId = taskId;
+        return pickedLocation;
+      } else {
+        throw new ISE("Unable to relocate task ([%s] -> [%s])", candidateLocation, pickedLocation);

Review Comment:
   should the slot be freed that was picked up? 



-- 
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: commits-unsubscribe@druid.apache.org

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


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org