Report runtime environment information in worker heartbeats

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java

@@ -10,6 +10,7 @@
 import io.temporal.api.enums.v1.TaskReachability;
 import io.temporal.api.history.v1.History;
 import io.temporal.api.history.v1.HistoryEvent;
+import io.temporal.api.worker.v1.EnvironmentInfo;
 import io.temporal.api.workflowservice.v1.*;
 import io.temporal.client.WorkflowInvocationHandler.InvocationType;
 import io.temporal.common.WorkflowExecutionHistory;
@@ -25,6 +26,7 @@
 import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import io.temporal.internal.sync.StubMarker;
 import io.temporal.internal.worker.HeartbeatManager;
+import io.temporal.internal.worker.WorkerEnvironmentInfo;
 import io.temporal.payload.storage.ExternalStorage;
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
@@ -58,6 +60,7 @@ final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClient
   private final WorkerFactoryRegistry workerFactoryRegistry = new WorkerFactoryRegistry();
   private final String workerGroupingKey = java.util.UUID.randomUUID().toString();
   private final @Nullable HeartbeatManager heartbeatManager;
+  private final @Nullable EnvironmentInfo workerEnvironmentInfo;
   private final @Nullable ExternalStorageRunner externalStorageRunner;
 
   /**
@@ -126,8 +129,11 @@ public static WorkflowClient newInstance(
     if (!heartbeatInterval.isNegative()) {
       this.heartbeatManager =
           new HeartbeatManager(workflowServiceStubs, options.getIdentity(), heartbeatInterval);
+      this.workerEnvironmentInfo =
+          options.isWorkerEnvironmentInfoDisabled() ? null : WorkerEnvironmentInfo.detect();
     } else {
       this.heartbeatManager = null;
+      this.workerEnvironmentInfo = null;
     }
   }
 
@@ -821,6 +827,12 @@ public HeartbeatManager getHeartbeatManager() {
     return heartbeatManager;
   }
 
+  @Override
+  @Nullable
+  public EnvironmentInfo getWorkerEnvironmentInfo() {
+    return workerEnvironmentInfo;
+  }
+
   @Override
   @Nullable
   public ExternalStorageRunner getExternalStorageRunner() {

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java

@@ -54,6 +54,7 @@ public static final class Builder {
     private QueryRejectCondition queryRejectCondition;
     private WorkflowClientPlugin[] plugins;
     private Duration workerHeartbeatInterval;
+    private boolean disableWorkerEnvironmentInfo;
     private ExternalStorage externalStorage;
 
     private Builder() {}
@@ -71,6 +72,7 @@ private Builder(WorkflowClientOptions options) {
       queryRejectCondition = options.queryRejectCondition;
       plugins = options.plugins;
       workerHeartbeatInterval = options.workerHeartbeatInterval;
+      disableWorkerEnvironmentInfo = options.disableWorkerEnvironmentInfo;
       externalStorage = options.externalStorage;
     }
 
@@ -187,6 +189,19 @@ public Builder setWorkerHeartbeatInterval(Duration workerHeartbeatInterval) {
       return this;
     }
 
+    /**
+     * Disables reporting the JVM version, detected hosting environments (Docker, Kubernetes, cloud
+     * platforms), and OS platform in worker heartbeats. This information is sent once per worker,
+     * with the first heartbeat accepted by the server.
+     *
+     * @param disableWorkerEnvironmentInfo true to omit environment information from heartbeats
+     */
+    @Experimental
+    public Builder setDisableWorkerEnvironmentInfo(boolean disableWorkerEnvironmentInfo) {
+      this.disableWorkerEnvironmentInfo = disableWorkerEnvironmentInfo;
+      return this;
+    }
+
     public WorkflowClientOptions build() {
       return new WorkflowClientOptions(
           namespace,
@@ -198,6 +213,7 @@ public WorkflowClientOptions build() {
           queryRejectCondition,
           plugins == null ? EMPTY_PLUGINS : plugins,
           resolveHeartbeatInterval(workerHeartbeatInterval),
+          disableWorkerEnvironmentInfo,
           externalStorage);
     }
 
@@ -226,6 +242,7 @@ public WorkflowClientOptions validateAndBuildWithDefaults() {
               : queryRejectCondition,
           plugins == null ? EMPTY_PLUGINS : plugins,
           resolveHeartbeatInterval(workerHeartbeatInterval),
+          disableWorkerEnvironmentInfo,
           externalStorage);
     }
 
@@ -269,6 +286,8 @@ private static Duration resolveHeartbeatInterval(Duration raw) {
 
   private final Duration workerHeartbeatInterval;
 
+  private final boolean disableWorkerEnvironmentInfo;
+
   private final @Nullable ExternalStorage externalStorage;
 
   private WorkflowClientOptions(
@@ -281,6 +300,7 @@ private WorkflowClientOptions(
       QueryRejectCondition queryRejectCondition,
       WorkflowClientPlugin[] plugins,
       Duration workerHeartbeatInterval,
+      boolean disableWorkerEnvironmentInfo,
       @Nullable ExternalStorage externalStorage) {
     this.namespace = namespace;
     this.dataConverter = dataConverter;
@@ -291,6 +311,7 @@ private WorkflowClientOptions(
     this.queryRejectCondition = queryRejectCondition;
     this.plugins = plugins;
     this.workerHeartbeatInterval = workerHeartbeatInterval;
+    this.disableWorkerEnvironmentInfo = disableWorkerEnvironmentInfo;
     this.externalStorage = externalStorage;
   }
 
@@ -365,6 +386,12 @@ public Duration getWorkerHeartbeatInterval() {
     return workerHeartbeatInterval;
   }
 
+  /** Returns true when runtime, hosting, and platform information is omitted from heartbeats. */
+  @Experimental
+  public boolean isWorkerEnvironmentInfoDisabled() {
+    return disableWorkerEnvironmentInfo;
+  }
+
   @Override
   public String toString() {
     return "WorkflowClientOptions{"
@@ -389,6 +416,8 @@ public String toString() {
         + Arrays.toString(plugins)
         + ", workerHeartbeatInterval="
         + workerHeartbeatInterval
+        + ", disableWorkerEnvironmentInfo="
+        + disableWorkerEnvironmentInfo
         + ", externalStorage="
         + externalStorage
         + '}';
@@ -409,6 +438,7 @@ public boolean equals(Object o) {
         && Arrays.equals(plugins, that.plugins)
         && com.google.common.base.Objects.equal(
             workerHeartbeatInterval, that.workerHeartbeatInterval)
+        && disableWorkerEnvironmentInfo == that.disableWorkerEnvironmentInfo
         && com.google.common.base.Objects.equal(externalStorage, that.externalStorage);
   }
 
@@ -424,6 +454,7 @@ public int hashCode() {
         queryRejectCondition,
         Arrays.hashCode(plugins),
         workerHeartbeatInterval,
+        disableWorkerEnvironmentInfo,
         externalStorage);
   }
 }

temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java

@@ -1,5 +1,6 @@
 package io.temporal.internal.client;
 
+import io.temporal.api.worker.v1.EnvironmentInfo;
 import io.temporal.client.WorkflowClient;
 import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import io.temporal.internal.worker.HeartbeatManager;
@@ -27,6 +28,13 @@ public interface WorkflowClientInternal {
   @Nullable
   HeartbeatManager getHeartbeatManager();
 
+  /**
+   * Environment information workers report in their heartbeats until the server accepts one, or
+   * null if disabled.
+   */
+  @Nullable
+  EnvironmentInfo getWorkerEnvironmentInfo();
+
   @Nullable
   ExternalStorageRunner getExternalStorageRunner();
 }

temporal-sdk/src/main/java/io/temporal/internal/worker/HeartbeatManager.java

@@ -37,6 +37,19 @@ public HeartbeatManager(WorkflowServiceStubs service, String identity, Duration
    */
   public void registerWorker(
       String namespace, String workerInstanceKey, Supplier<WorkerHeartbeat> callback) {
+    registerWorker(namespace, workerInstanceKey, callback, () -> {});
+  }
+
+  /**
+   * @param onHeartbeatAccepted invoked, from the heartbeat thread, each time a heartbeat produced
+   *     by {@code callback} has been accepted by the server
+   */
+  public void registerWorker(
+      String namespace,
+      String workerInstanceKey,
+      Supplier<WorkerHeartbeat> callback,
+      Runnable onHeartbeatAccepted) {
+    WorkerCallbacks callbacks = new WorkerCallbacks(callback, onHeartbeatAccepted);
     synchronized (lock) {
       if (unimplementedNamespaces.contains(namespace)) {
         return;
@@ -45,12 +58,12 @@ public void registerWorker(
           namespace,
           (ns, existing) -> {
             if (existing != null && !existing.isShutdown()) {
-              existing.registerWorker(workerInstanceKey, callback);
+              existing.registerWorker(workerInstanceKey, callbacks);
               return existing;
             }
             SharedNamespaceWorker nsWorker =
                 new SharedNamespaceWorker(this, service, ns, identity, interval);
-            nsWorker.registerWorker(workerInstanceKey, callback);
+            nsWorker.registerWorker(workerInstanceKey, callbacks);
             return nsWorker;
           });
     }
@@ -96,6 +109,16 @@ void markNamespaceUnimplemented(String namespace) {
     }
   }
 
+  private static final class WorkerCallbacks {
+    final Supplier<WorkerHeartbeat> heartbeat;
+    final Runnable heartbeatAccepted;
+
+    WorkerCallbacks(Supplier<WorkerHeartbeat> heartbeat, Runnable heartbeatAccepted) {
+      this.heartbeat = heartbeat;
+      this.heartbeatAccepted = heartbeatAccepted;
+    }
+  }
+
   /**
    * Handles heartbeating for all workers in a specific namespace. Each instance owns its own
    * scheduler thread and callback map.
@@ -105,8 +128,7 @@ static class SharedNamespaceWorker {
     private final WorkflowServiceStubs service;
     private final String namespace;
     private final String identity;
-    private final ConcurrentHashMap<String, Supplier<WorkerHeartbeat>> callbacks =
-        new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, WorkerCallbacks> callbacks = new ConcurrentHashMap<>();
     private final ScheduledExecutorService scheduler;
 
     SharedNamespaceWorker(
@@ -130,8 +152,8 @@ static class SharedNamespaceWorker {
           this::heartbeatTick, 0, interval.toMillis(), TimeUnit.MILLISECONDS);
     }
 
-    void registerWorker(String workerInstanceKey, Supplier<WorkerHeartbeat> callback) {
-      callbacks.put(workerInstanceKey, callback);
+    void registerWorker(String workerInstanceKey, WorkerCallbacks workerCallbacks) {
+      callbacks.put(workerInstanceKey, workerCallbacks);
     }
 
     void unregisterWorker(String workerInstanceKey) {
@@ -165,9 +187,11 @@ private void heartbeatTick() {
       if (callbacks.isEmpty()) return;
 
       List<WorkerHeartbeat> heartbeats = new ArrayList<>();
-      for (Map.Entry<String, Supplier<WorkerHeartbeat>> entry : callbacks.entrySet()) {
+      List<Runnable> acceptedCallbacks = new ArrayList<>();
+      for (Map.Entry<String, WorkerCallbacks> entry : callbacks.entrySet()) {
         try {
-          heartbeats.add(entry.getValue().get());
+          heartbeats.add(entry.getValue().heartbeat.get());
+          acceptedCallbacks.add(entry.getValue().heartbeatAccepted);
         } catch (Exception e) {
           log.warn(
               "Failed to build heartbeat for worker {} in namespace {}",
@@ -196,8 +220,18 @@ private void heartbeatTick() {
           return;
         }
         log.warn("Failed to send worker heartbeat for namespace {}", namespace, e);
+        return;
       } catch (Exception e) {
         log.warn("Failed to send worker heartbeat for namespace {}", namespace, e);
+        return;
+      }
+
+      for (Runnable accepted : acceptedCallbacks) {
+        try {
+          accepted.run();
+        } catch (Exception e) {
+          log.warn("Heartbeat accepted callback failed in namespace {}", namespace, e);
+        }
       }
     }
   }