Poll API, health management for recorders, backend daemon for reporting load, aggregation window management, work assignment scheduling

aggregation/src/main/java/fk/prof/aggregation/model/FinalizedAggregationWindow.java

@@ -12,22 +12,22 @@ public class FinalizedAggregationWindow {
   private final String clusterId;
   private final String procId;
   private final LocalDateTime start;
-  private final LocalDateTime endWithTolerance;
+  private final LocalDateTime endedAt;
   protected final Map<Long, FinalizedProfileWorkInfo> workInfoLookup;
   protected final FinalizedCpuSamplingAggregationBucket cpuSamplingAggregationBucket;
 
   public FinalizedAggregationWindow(String appId,
                                     String clusterId,
                                     String procId,
                                     LocalDateTime start,
-                                    LocalDateTime endWithTolerance,
+                                    LocalDateTime endedAt,
                                     Map<Long, FinalizedProfileWorkInfo> workInfoLookup,
                                     FinalizedCpuSamplingAggregationBucket cpuSamplingAggregationBucket) {
     this.appId = appId;
     this.clusterId = clusterId;
     this.procId = procId;
     this.start = start;
-    this.endWithTolerance = endWithTolerance;
+    this.endedAt = endedAt;
     this.workInfoLookup = workInfoLookup;
     this.cpuSamplingAggregationBucket = cpuSamplingAggregationBucket;
   }
@@ -36,6 +36,11 @@ public FinalizedProfileWorkInfo getDetailsForWorkId(long workId) {
     return this.workInfoLookup.get(workId);
   }
 
+  //NOTE: This is computed on expiry of aggregation window, null otherwise. Having a getter here to make this testable
+  public LocalDateTime getEndedAt() {
+    return this.endedAt;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (o == this) {
@@ -50,7 +55,7 @@ public boolean equals(Object o) {
         && this.clusterId.equals(other.clusterId)
         && this.procId.equals(other.procId)
         && this.start.equals(other.start)
-        && this.endWithTolerance.equals(other.endWithTolerance)
+        && this.endedAt == null ? other.endedAt == null : this.endedAt.equals(other.endedAt)
         && this.workInfoLookup.equals(other.workInfoLookup)
         && this.cpuSamplingAggregationBucket.equals(other.cpuSamplingAggregationBucket);
   }
@@ -59,7 +64,7 @@ protected Header buildHeaderProto(int version, WorkType workType) {
     return Header.newBuilder()
         .setFormatVersion(version)
         .setWorkType(workType)
-        .setAggregationEndTime(endWithTolerance.atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME))
+        .setAggregationEndTime(endedAt == null ? null : endedAt.atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME))
         .setAggregationStartTime(start.atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME))
         .setAppId(appId)
         .setClusterId(clusterId)

aggregation/src/main/java/fk/prof/aggregation/model/FinalizedProfileWorkInfo.java

@@ -7,6 +7,9 @@
 import java.time.temporal.ChronoUnit;
 import java.util.Map;
 
+/**
+ * TODO: Add recorder info(aggregated proto) to this class, accept it in constructor which is called from ProfileWorkInfo#buildFinalizedEntity method
+ */
 public class FinalizedProfileWorkInfo {
   private final int recorderVersion;
   private final AggregationState state;
@@ -40,6 +43,10 @@ public LocalDateTime getEndedAt() {
     return endedAt;
   }
 
+  public AggregationState getState() {
+    return state;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (o == this) {

backend/src/idl/backend.proto

@@ -4,6 +4,49 @@ option java_outer_classname = "BackendDTO";
 
 message LoadReportRequest {
     required string ip = 1;
-    required float load = 2;
-    required int64 curr_tick = 3;
+    required uint32 port = 2;
+    required float load = 3;
+    required int64 curr_tick = 4;
+}
+
+message WorkProfile {
+    required uint32 duration = 1;
+    required uint32 coverage_pct = 2;
+    repeated Work work = 3;
+    required string description = 4;
+}
+
+enum WorkType {
+    cpu_sample_work = 0;
+    thread_sample_work = 1;
+    monitor_contention_work = 2;
+    monitor_wait_work = 3;
+}
+
+message Work {
+    required WorkType w_type = 1;
+    optional CpuSampleWork cpu_sample = 2;
+    optional ThreadSampleWork thd_sample = 3;
+    optional MonitorContentionWork monitor_block = 4;
+    optional MonitorWaitWork monitor_wait = 5;
+}
+
+message CpuSampleWork {
+    required uint32 frequency = 1;
+    required uint32 max_frames = 2;
+}
+
+message ThreadSampleWork {
+    required uint32 frequency = 1;
+    required uint32 max_frames = 2;
+}
+
+message MonitorContentionWork {
+    required uint32 max_monitors = 1;
+    required uint32 max_frames = 2;
+}
+
+message MonitorWaitWork {
+    required uint32 max_monitors = 1;
+    required uint32 max_frames = 2;
 }

backend/src/main/conf/local.json

@@ -1,7 +1,11 @@
 {
+  "ip.address": "127.0.0.1",
+  "backend.version": 1,
   "backend.http.port": 2491,
   "leader.http.port": 2496,
   "load.report.interval.secs": 60,
+  "recorder.defunct.threshold.secs": 120,
+  "max.simultaneous.profiles": 100,
   "backend.http.server": {
     "idle.timeout.secs": 120
   },
@@ -16,7 +20,7 @@
     "compression": true
   },
   "vertxOptions": {
-    "blockedThreadCheckInterval": 1000
+    "blockedThreadCheckInterval": 1000000
   },
   "backendHttpOptions": {
     "config": {
@@ -41,11 +45,22 @@
       "aggregation.enabled": false,
       "leader.watching.path": "/backends",
       "leader.mutex.path": "/leader",
-      "leader.sleep.ms": 60000
+      "leader.sleep.ms": 60000,
+      "kill.behavior": "KILL"
     }
   },
   "leaderHttpOptions": {
     "backend.association.path": "/association",
     "load.miss.tolerance": 1
+  },
+  "daemonOptions": {
+    "worker": true,
+    "config": {
+      "aggregation.window.duration.mins": 30,
+      "aggregation.window.end.tolerance.secs": 120,
+      "workprofile.refresh.offset.secs": 300,
+      "scheduling.buffer.secs": 30,
+      "work.assignment.max.delay.secs": 120
+    }
   }
 }

backend/src/main/java/fk/prof/backend/BackendApplication.java

@@ -1,8 +1,13 @@
 package fk.prof.backend;
 
+import fk.prof.backend.worker.BackendDaemon;
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
 import org.apache.commons.cli.*;
 
 public class BackendApplication {
+  private static Logger logger = LoggerFactory.getLogger(BackendDaemon.class);
+
   public static void main(String[] args) throws Exception {
     ConfigManager.setDefaultSystemProperties();
     CommandLineParser parser = new DefaultParser();
@@ -20,6 +25,12 @@ public static void main(String[] args) throws Exception {
     String confPath = cmd.getOptionValue("c");
 
     BackendManager backendManager = new BackendManager(confPath);
-    backendManager.launch();
+    backendManager.launch().setHandler(ar -> {
+      if(ar.succeeded())  {
+        logger.info("Backend launched");
+      } else {
+        logger.error("Error launching backend: ", ar.cause());
+      }
+    });
   }
 }

backend/src/main/java/fk/prof/backend/BackendManager.java

@@ -2,16 +2,19 @@
 
 import com.google.common.base.Preconditions;
 import fk.prof.backend.deployer.VerticleDeployer;
-import fk.prof.backend.deployer.impl.BackendHttpVerticleDeployer;
-import fk.prof.backend.deployer.impl.LeaderElectionParticipatorVerticleDeployer;
-import fk.prof.backend.deployer.impl.LeaderElectionWatcherVerticleDeployer;
-import fk.prof.backend.deployer.impl.LeaderHttpVerticleDeployer;
+import fk.prof.backend.deployer.impl.*;
 import fk.prof.backend.leader.election.LeaderElectedTask;
+import fk.prof.backend.model.aggregation.AggregationWindowLookupStore;
+import fk.prof.backend.model.assignment.ProcessGroupAssociationStore;
+import fk.prof.backend.model.assignment.SimultaneousWorkAssignmentCounter;
+import fk.prof.backend.model.assignment.impl.ProcessGroupAssociationStoreImpl;
+import fk.prof.backend.model.assignment.impl.SimultaneousWorkAssignmentCounterImpl;
 import fk.prof.backend.model.association.BackendAssociationStore;
 import fk.prof.backend.model.association.ProcessGroupCountBasedBackendComparator;
 import fk.prof.backend.model.association.impl.ZookeeperBasedBackendAssociationStore;
 import fk.prof.backend.model.election.impl.InMemoryLeaderStore;
-import fk.prof.backend.service.ProfileWorkService;
+import fk.prof.backend.model.aggregation.impl.AggregationWindowLookupStoreImpl;
+import fk.prof.backend.model.policy.PolicyStore;
 import io.vertx.core.*;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.logging.Logger;
@@ -23,8 +26,10 @@
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /**
  * TODO: Deployment process is liable to changes later
@@ -76,16 +81,24 @@ public Future<Void> close() {
   public Future<Void> launch() {
     Future result = Future.future();
     InMemoryLeaderStore leaderStore = new InMemoryLeaderStore(configManager.getIPAddress());
-    ProfileWorkService profileWorkService = new ProfileWorkService();
-
-    VerticleDeployer backendHttpVerticleDeployer = new BackendHttpVerticleDeployer(vertx, configManager, leaderStore, profileWorkService);
-    backendHttpVerticleDeployer.deploy().setHandler(backendDeployResult -> {
+    AggregationWindowLookupStore aggregationWindowLookupStore = new AggregationWindowLookupStoreImpl();
+    ProcessGroupAssociationStore processGroupAssociationStore = new ProcessGroupAssociationStoreImpl(configManager.getRecorderDefunctThresholdInSeconds());
+    SimultaneousWorkAssignmentCounter simultaneousWorkAssignmentCounter = new SimultaneousWorkAssignmentCounterImpl(configManager.getMaxSimultaneousProfiles());
+
+    VerticleDeployer backendHttpVerticleDeployer = new BackendHttpVerticleDeployer(vertx, configManager, leaderStore, aggregationWindowLookupStore, processGroupAssociationStore);
+    VerticleDeployer backendDaemonVerticleDeployer = new BackendDaemonVerticleDeployer(vertx, configManager, leaderStore, processGroupAssociationStore, aggregationWindowLookupStore, simultaneousWorkAssignmentCounter);
+    CompositeFuture backendDeploymentFuture = CompositeFuture.all(backendHttpVerticleDeployer.deploy(), backendDaemonVerticleDeployer.deploy());
+    backendDeploymentFuture.setHandler(backendDeployResult -> {
       if (backendDeployResult.succeeded()) {
         try {
-          List<String> backendDeployments = backendDeployResult.result().list();
+          List<String> backendDeployments = backendDeployResult.result().list().stream()
+              .flatMap(fut -> ((CompositeFuture)fut).list().stream())
+              .map(deployment -> (String)deployment)
+              .collect(Collectors.toList());
 
           BackendAssociationStore backendAssociationStore = createBackendAssociationStore(vertx, curatorClient);
-          VerticleDeployer leaderHttpVerticleDeployer = new LeaderHttpVerticleDeployer(vertx, configManager, backendAssociationStore);
+          PolicyStore policyStore = new PolicyStore();
+          VerticleDeployer leaderHttpVerticleDeployer = new LeaderHttpVerticleDeployer(vertx, configManager, backendAssociationStore, policyStore);
           Runnable leaderElectedTask = createLeaderElectedTask(vertx, leaderHttpVerticleDeployer, backendDeployments);
 
           VerticleDeployer leaderElectionParticipatorVerticleDeployer = new LeaderElectionParticipatorVerticleDeployer(

backend/src/main/java/fk/prof/backend/ConfigManager.java

@@ -15,17 +15,21 @@
  */
 public class ConfigManager {
   private static final String IP_ADDRESS_KEY = "ip.address";
+  private static final String BACKEND_VERSION_KEY = "backend.version";
   private static final String BACKEND_HTTP_PORT_KEY = "backend.http.port";
   private static final String LEADER_HTTP_PORT_KEY = "leader.http.port";
   private static final String BACKEND_HTTP_SERVER_OPTIONS_KEY = "backend.http.server";
   private static final String LEADER_HTTP_SERVER_OPTIONS_KEY = "leader.http.server";
   private static final String HTTP_CLIENT_OPTIONS_KEY = "http.client";
   private static final String LOAD_REPORT_INTERVAL_KEY = "load.report.interval.secs";
+  private static final String RECORDER_DEFUNCT_THRESHOLD_KEY = "recorder.defunct.threshold.secs";
+  private static final String MAX_SIMULTANEOUS_PROFILES_KEY = "max.simultaneous.profiles";
   private static final String VERTX_OPTIONS_KEY = "vertxOptions";
   private static final String BACKEND_HTTP_DEPLOYMENT_OPTIONS_KEY = "backendHttpOptions";
   private static final String CURATOR_OPTIONS_KEY = "curatorOptions";
   private static final String LEADER_ELECTION_DEPLOYMENT_OPTIONS_KEY = "leaderElectionOptions";
   private static final String LEADER_HTTP_DEPLOYMENT_OPTIONS_KEY = "leaderHttpOptions";
+  private static final String BACKEND_DAEMON_OPTIONS_KEY = "daemonOptions";
   private static final String LOGFACTORY_SYSTEM_PROPERTY_KEY = "vertx.logger-delegate-factory-class-name";
   private static final String LOGFACTORY_SYSTEM_PROPERTY_DEFAULT_VALUE = "io.vertx.core.logging.SLF4JLogDelegateFactory";
 
@@ -48,6 +52,10 @@ public String getIPAddress() {
     return config.getString(IP_ADDRESS_KEY, "127.0.0.1");
   }
 
+  public int getBackendVersion() {
+    return config.getInteger(BACKEND_VERSION_KEY);
+  }
+
   public int getBackendHttpPort() {
     return config.getInteger(BACKEND_HTTP_PORT_KEY, 2491);
   }
@@ -60,6 +68,14 @@ public int getLoadReportIntervalInSeconds() {
     return config.getInteger(LOAD_REPORT_INTERVAL_KEY, 60);
   }
 
+  public int getRecorderDefunctThresholdInSeconds() {
+    return config.getInteger(RECORDER_DEFUNCT_THRESHOLD_KEY, 120);
+  }
+
+  public int getMaxSimultaneousProfiles() {
+    return config.getInteger(MAX_SIMULTANEOUS_PROFILES_KEY, 10);
+  }
+
   public JsonObject getBackendHttpServerConfig() {
     return config.getJsonObject(BACKEND_HTTP_SERVER_OPTIONS_KEY, new JsonObject());
   }
@@ -92,6 +108,10 @@ public JsonObject getLeaderHttpDeploymentConfig() {
     return enrichDeploymentConfig(config.getJsonObject(LEADER_HTTP_DEPLOYMENT_OPTIONS_KEY, new JsonObject()));
   }
 
+  public JsonObject getBackendDaemonDeploymentConfig() {
+    return enrichDeploymentConfig(config.getJsonObject(BACKEND_DAEMON_OPTIONS_KEY, new JsonObject()));
+  }
+
   private JsonObject enrichDeploymentConfig(JsonObject deploymentConfig) {
     if(deploymentConfig.getJsonObject("config") == null) {
       deploymentConfig.put("config", new JsonObject());