diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java index c92d1d8390..0b892425e0 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java +++ b/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() { diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java index e0e6a5a7b6..3015d20233 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java +++ b/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); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java index 982ae56724..90017cd575 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java +++ b/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(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/HeartbeatManager.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/HeartbeatManager.java index 09edd173e3..b40f9d95d5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/HeartbeatManager.java +++ b/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 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 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 heartbeat; + final Runnable heartbeatAccepted; + + WorkerCallbacks(Supplier 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> callbacks = - new ConcurrentHashMap<>(); + private final ConcurrentHashMap 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 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 heartbeats = new ArrayList<>(); - for (Map.Entry> entry : callbacks.entrySet()) { + List acceptedCallbacks = new ArrayList<>(); + for (Map.Entry 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); + } } } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java new file mode 100644 index 0000000000..9fcaf1ba71 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java @@ -0,0 +1,312 @@ +package io.temporal.internal.worker; + +import io.temporal.api.worker.v1.EnvironmentInfo; +import io.temporal.api.worker.v1.EnvironmentInfo.Architecture; +import io.temporal.api.worker.v1.EnvironmentInfo.HostingEnvironment; +import io.temporal.api.worker.v1.EnvironmentInfo.HostingEnvironment.HostingEnvironmentType; +import io.temporal.api.worker.v1.EnvironmentInfo.LinuxPlatform; +import io.temporal.api.worker.v1.EnvironmentInfo.MacOSPlatform; +import io.temporal.api.worker.v1.EnvironmentInfo.Platform; +import io.temporal.api.worker.v1.EnvironmentInfo.Runtime; +import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType; +import io.temporal.api.worker.v1.EnvironmentInfo.WindowsPlatform; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.function.Function; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Detects the runtime, hosting environment, and platform information reported in the first accepted + * worker heartbeat. + */ +public final class WorkerEnvironmentInfo { + private static final Logger log = LoggerFactory.getLogger(WorkerEnvironmentInfo.class); + + private WorkerEnvironmentInfo() {} + + /** + * Never throws: this runs during client creation, and telemetry must not break it. System + * property, environment, and filesystem access can all fail under a security manager, in which + * case whatever was collected before the failure is returned. + */ + public static EnvironmentInfo detect() { + EnvironmentInfo.Builder builder = EnvironmentInfo.newBuilder(); + try { + builder.addRuntimes( + Runtime.newBuilder() + .setType(RuntimeType.RUNTIME_TYPE_JVM) + .setVersion(nullToEmpty(System.getProperty("java.version")))); + builder.addAllHostingEnvironments(detectHostingEnvironments(System::getenv)); + Platform platform = detectPlatform(); + if (platform != null) { + builder.setPlatform(platform); + } + } catch (RuntimeException e) { + log.warn("Failed to detect worker environment information, reporting partial results", e); + } + return builder.build(); + } + + /** + * Several environments may be detected at once, e.g. Docker inside Kubernetes or Azure Functions + * inside Azure App Service. + */ + static List detectHostingEnvironments(Function env) { + List environments = new ArrayList<>(); + if (isDocker()) { + environments.add( + hostingEnvironment(HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_DOCKER, "")); + } + if (hasAnyEnv(env, "KUBERNETES_SERVICE_HOST")) { + environments.add(hostingEnvironment(HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_K8S, "")); + } + if (hasAnyEnv(env, "AWS_LAMBDA_FUNCTION_NAME")) { + environments.add( + hostingEnvironment(HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AWS_LAMBDA, "")); + } + if (hasAnyEnv(env, "ECS_CONTAINER_METADATA_URI_V4", "ECS_CONTAINER_METADATA_URI")) { + environments.add( + hostingEnvironment(HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AWS_ECS, "")); + } + if (hasAnyEnv(env, "K_SERVICE", "CLOUD_RUN_JOB", "CLOUD_RUN_WORKER_POOL")) { + environments.add( + hostingEnvironment(HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_GOOGLE_CLOUD_RUN, "")); + } + if (hasAnyEnv(env, "GAE_SERVICE")) { + environments.add( + hostingEnvironment( + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_GOOGLE_APP_ENGINE, "")); + } + if (hasAnyEnv(env, "WEBSITE_SITE_NAME")) { + environments.add( + hostingEnvironment( + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AZURE_APP_SERVICE, + envValue(env, "WEBSITE_PLATFORM_VERSION"))); + } + String functionsVersion = envValue(env, "FUNCTIONS_EXTENSION_VERSION"); + if (!functionsVersion.isEmpty()) { + environments.add( + hostingEnvironment( + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AZURE_FUNCTIONS, functionsVersion)); + } + if (hasAnyEnv(env, "CONTAINER_APP_NAME", "CONTAINER_APP_JOB_NAME")) { + environments.add( + hostingEnvironment( + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AZURE_CONTAINER_APPS, "")); + } + return environments; + } + + private static HostingEnvironment hostingEnvironment( + HostingEnvironmentType type, String version) { + return HostingEnvironment.newBuilder().setType(type).setVersion(version).build(); + } + + private static String envValue(Function env, String name) { + return nullToEmpty(env.apply(name)).trim(); + } + + private static boolean hasAnyEnv(Function env, String... names) { + for (String name : names) { + if (!envValue(env, name).isEmpty()) { + return true; + } + } + return false; + } + + private static boolean isDocker() { + if (isWindows(System.getProperty("os.name"))) { + return false; + } + if (Files.exists(Paths.get("/.dockerenv"))) { + return true; + } + Path cgroup = Paths.get("/proc/self/cgroup"); + if (!Files.isReadable(cgroup)) { + return false; + } + try { + return cgroupsIndicateDocker(Files.readAllLines(cgroup, StandardCharsets.UTF_8)); + } catch (IOException | RuntimeException e) { + return false; + } + } + + /** + * Reports whether any cgroup path has a {@code docker} or {@code docker-.scope} component. + */ + static boolean cgroupsIndicateDocker(List cgroupLines) { + for (String line : cgroupLines) { + int idx = line.lastIndexOf(':'); + String path = idx >= 0 ? line.substring(idx + 1) : line; + for (String component : path.split("/")) { + if (component.equals("docker") + || (component.startsWith("docker-") && component.endsWith(".scope"))) { + return true; + } + } + } + return false; + } + + @Nullable + private static Platform detectPlatform() { + String osName = nullToEmpty(System.getProperty("os.name")); + String name = osName.toLowerCase(Locale.ROOT); + String osVersion = nullToEmpty(System.getProperty("os.version")); + Architecture architecture = detectArchitecture(); + if (name.contains("linux")) { + return Platform.newBuilder() + .setLinux( + LinuxPlatform.newBuilder() + .setVersion(linuxVersion(osVersion)) + .setArchitecture(architecture)) + .build(); + } + if (name.contains("mac") || name.contains("darwin")) { + return Platform.newBuilder() + .setMacos(MacOSPlatform.newBuilder().setVersion(osVersion).setArchitecture(architecture)) + .build(); + } + if (isWindows(name)) { + return Platform.newBuilder() + .setWindows( + WindowsPlatform.newBuilder() + .setVersion(windowsVersion(osName, osVersion)) + .setArchitecture(architecture) + .setCrt( + windowsCrt( + javaMajorVersion(System.getProperty("java.specification.version"))))) + .build(); + } + return null; + } + + static Architecture detectArchitecture() { + switch (nullToEmpty(System.getProperty("os.arch")).toLowerCase(Locale.ROOT)) { + case "amd64": + case "x86_64": + return Architecture.ARCHITECTURE_AMD64; + case "aarch64": + case "arm64": + return Architecture.ARCHITECTURE_ARM64; + default: + return Architecture.ARCHITECTURE_UNSPECIFIED; + } + } + + /** + * Windows JDKs before 11 were built with Visual Studio toolchains that ship their own MSVC + * runtime; JDK 11 onward links against the Universal CRT. + */ + private static WindowsPlatform.Crt windowsCrt(int javaMajor) { + if (javaMajor <= 0) { + return WindowsPlatform.Crt.CRT_UNSPECIFIED; + } + return javaMajor >= 11 ? WindowsPlatform.Crt.CRT_UCRT : WindowsPlatform.Crt.CRT_MSVCRT; + } + + static int javaMajorVersion(@Nullable String specificationVersion) { + String version = nullToEmpty(specificationVersion); + if (version.startsWith("1.")) { + version = version.substring(2); + } + int end = version.indexOf('.'); + if (end >= 0) { + version = version.substring(0, end); + } + try { + return Integer.parseInt(version); + } catch (NumberFormatException e) { + return 0; + } + } + + /** + * Windows 11 still reports {@code os.version} as {@code 10.0}; since JDK 17.0.1 {@code os.name} + * carries the marketing version ("Windows 11"), so prefer it when it is a plain number that + * {@code os.version} contradicts. Server editions ("Windows Server 2022") are left as-is because + * their year is not a version. + */ + static String windowsVersion(String osName, String osVersion) { + String prefix = "windows "; + String lower = osName.toLowerCase(Locale.ROOT); + if (!lower.startsWith(prefix) || lower.startsWith("windows server")) { + return osVersion; + } + String marketing = osName.substring(prefix.length()).trim(); + int marketingMajor = leadingInt(marketing); + if (marketingMajor <= 0 + || !marketing.matches("[0-9]+(\\.[0-9]+)*") + || marketingMajor <= leadingInt(osVersion)) { + return osVersion; + } + return marketing; + } + + private static int leadingInt(String version) { + int end = 0; + while (end < version.length() && Character.isDigit(version.charAt(end))) { + end++; + } + try { + return Integer.parseInt(version.substring(0, end)); + } catch (NumberFormatException e) { + return 0; + } + } + + private static boolean isWindows(@Nullable String osName) { + return nullToEmpty(osName).toLowerCase(Locale.ROOT).contains("windows"); + } + + /** + * The JVM only exposes the kernel release as {@code os.version}; prefer the distribution version + * from os-release to match what Core reports. + */ + private static String linuxVersion(String kernelVersion) { + for (String path : new String[] {"/etc/os-release", "/usr/lib/os-release"}) { + String version = osReleaseValue(Paths.get(path), "VERSION_ID"); + if (!version.isEmpty()) { + return version; + } + } + return kernelVersion; + } + + private static String osReleaseValue(Path path, String key) { + if (!Files.isReadable(path)) { + return ""; + } + try { + for (String line : Files.readAllLines(path, StandardCharsets.UTF_8)) { + int idx = line.indexOf('='); + if (idx > 0 && line.substring(0, idx).equals(key)) { + String value = line.substring(idx + 1).trim(); + if (value.length() >= 2 + && (value.startsWith("\"") && value.endsWith("\"") + || value.startsWith("'") && value.endsWith("'"))) { + value = value.substring(1, value.length() - 1); + } + return value; + } + } + } catch (IOException | RuntimeException e) { + // Fall through to the caller's fallback. + } + return ""; + } + + private static String nullToEmpty(@Nullable String value) { + return value == null ? "" : value; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 818308aace..6627e3fd99 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -8,6 +8,7 @@ import io.temporal.api.deployment.v1.WorkerDeploymentVersion; import io.temporal.api.enums.v1.TaskQueueType; import io.temporal.api.enums.v1.WorkerStatus; +import io.temporal.api.worker.v1.EnvironmentInfo; import io.temporal.api.worker.v1.PluginInfo; import io.temporal.api.worker.v1.WorkerHeartbeat; import io.temporal.api.worker.v1.WorkerHostInfo; @@ -79,6 +80,9 @@ public final class Worker { private final @Nonnull WorkflowExecutorCache cache; private final Map previousHeartbeatSnapshots = new ConcurrentHashMap<>(); private volatile Supplier heartbeatSupplier; + // Reported in every heartbeat (including the one embedded in ShutdownWorkerRequest) until the + // server accepts one, then cleared so it is sent only once per worker. + private final AtomicReference pendingEnvironmentInfo = new AtomicReference<>(); private static final class TaskSnapshot { final int processed; @@ -595,7 +599,14 @@ List getActiveTaskQueueTypes() { return types; } - Supplier buildHeartbeatCallback(String workerGroupingKey) { + /** Called by the heartbeat manager once a heartbeat produced by this worker was accepted. */ + void onHeartbeatAccepted() { + pendingEnvironmentInfo.set(null); + } + + Supplier buildHeartbeatCallback( + String workerGroupingKey, @Nullable EnvironmentInfo environmentInfo) { + pendingEnvironmentInfo.set(environmentInfo); // The callback can be invoked concurrently from the heartbeat scheduler and the shutdown path final Object callbackLock = new Object(); final AtomicReference lastHeartbeatTime = new AtomicReference<>(null); @@ -630,6 +641,11 @@ Supplier buildHeartbeatCallback(String workerGroupingKey) { } lastHeartbeatTime.set(now); + EnvironmentInfo pendingEnvironment = pendingEnvironmentInfo.get(); + if (pendingEnvironment != null) { + hb.setEnvironment(pendingEnvironment); + } + // Deployment version if (options.getDeploymentOptions() != null && options.getDeploymentOptions().getVersion() != null) { diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java index 70bcf28c76..8ec3adba40 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java @@ -4,6 +4,7 @@ import com.google.common.base.Preconditions; import com.google.common.base.Strings; import com.uber.m3.tally.Scope; +import io.temporal.api.worker.v1.EnvironmentInfo; import io.temporal.api.worker.v1.WorkerHeartbeat; import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest; import io.temporal.api.workflowservice.v1.DescribeNamespaceResponse; @@ -327,10 +328,15 @@ private void doStart() { // Register heartbeat callbacks after workers are started. if (hbManager != null && namespaceCapabilities.isWorkerHeartbeats()) { + EnvironmentInfo environmentInfo = clientInternal.getWorkerEnvironmentInfo(); for (Worker worker : workers.values()) { Supplier heartbeatSupplier = - worker.buildHeartbeatCallback(workerGroupingKey); - hbManager.registerWorker(namespace, worker.getWorkerInstanceKey(), heartbeatSupplier); + worker.buildHeartbeatCallback(workerGroupingKey, environmentInfo); + hbManager.registerWorker( + namespace, + worker.getWorkerInstanceKey(), + heartbeatSupplier, + worker::onHeartbeatAccepted); worker.setHeartbeatSupplier(heartbeatSupplier); } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/HeartbeatManagerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/HeartbeatManagerTest.java index 96192e4c0c..7cffb0cd5b 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/HeartbeatManagerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/HeartbeatManagerTest.java @@ -4,10 +4,13 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; +import io.temporal.api.worker.v1.EnvironmentInfo; import io.temporal.api.worker.v1.WorkerHeartbeat; import io.temporal.api.workflowservice.v1.*; import io.temporal.serviceclient.WorkflowServiceStubs; import java.time.Duration; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -222,6 +225,87 @@ public void testNamespaceSchedulerStopsWhenLastWorkerUnregisters() throws Except verify(blockingStub, after(VERIFY_TIMEOUT_MS).never()).recordWorkerHeartbeat(any()); } + @Test + public void testEnvironmentInfoSentUntilAccepted() throws Exception { + EnvironmentInfo environment = + EnvironmentInfo.newBuilder() + .addRuntimes( + EnvironmentInfo.Runtime.newBuilder() + .setType(EnvironmentInfo.Runtime.RuntimeType.RUNTIME_TYPE_JVM) + .setVersion("17")) + .build(); + // Fail the first delivery so the environment must be retried. + when(blockingStub.recordWorkerHeartbeat(any())) + .thenThrow(new io.grpc.StatusRuntimeException(io.grpc.Status.UNAVAILABLE)) + .thenReturn(RecordWorkerHeartbeatResponse.getDefaultInstance()); + + manager = new HeartbeatManager(service, "test-identity", FAST_INTERVAL); + + // Mirrors Worker.buildHeartbeatCallback: the environment is embedded by the supplier until the + // accepted callback clears it. + AtomicReference pending = new AtomicReference<>(environment); + manager.registerWorker( + "default", + "worker-1", + () -> { + WorkerHeartbeat.Builder hb = + WorkerHeartbeat.newBuilder().setWorkerInstanceKey("worker-1"); + EnvironmentInfo env = pending.get(); + if (env != null) { + hb.setEnvironment(env); + } + return hb.build(); + }, + () -> pending.set(null)); + + verify(blockingStub, timeout(VERIFY_TIMEOUT_MS).atLeast(3)).recordWorkerHeartbeat(any()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(RecordWorkerHeartbeatRequest.class); + verify(blockingStub, atLeast(3)).recordWorkerHeartbeat(captor.capture()); + + List requests = captor.getAllValues(); + assertEquals(environment, requests.get(0).getWorkerHeartbeat(0).getEnvironment()); + assertEquals(environment, requests.get(1).getWorkerHeartbeat(0).getEnvironment()); + for (RecordWorkerHeartbeatRequest request : requests.subList(2, requests.size())) { + assertFalse(request.getWorkerHeartbeat(0).hasEnvironment()); + } + } + + @Test + public void testAcceptedCallbackNotInvokedOnFailure() throws Exception { + when(blockingStub.recordWorkerHeartbeat(any())) + .thenThrow(new io.grpc.StatusRuntimeException(io.grpc.Status.UNAVAILABLE)) + .thenThrow(new RuntimeException("boom")) + .thenReturn(RecordWorkerHeartbeatResponse.getDefaultInstance()); + manager = new HeartbeatManager(service, "test-identity", FAST_INTERVAL); + + WorkerHeartbeat hb = WorkerHeartbeat.newBuilder().setWorkerInstanceKey("worker-1").build(); + Runnable accepted = mock(Runnable.class); + manager.registerWorker("default", "worker-1", () -> hb, accepted); + + verify(blockingStub, timeout(VERIFY_TIMEOUT_MS).atLeast(3)).recordWorkerHeartbeat(any()); + verify(accepted, timeout(VERIFY_TIMEOUT_MS).atLeastOnce()).run(); + // Two failed RPCs preceded the first success, so there must be fewer acceptances than RPCs. + assertTrue( + mockingDetails(accepted).getInvocations().size() + <= mockingDetails(blockingStub).getInvocations().size() - 2); + } + + @Test + public void testEnvironmentInfoOmittedWhenNull() throws Exception { + manager = new HeartbeatManager(service, "test-identity", FAST_INTERVAL); + + WorkerHeartbeat hb = WorkerHeartbeat.newBuilder().setWorkerInstanceKey("worker-1").build(); + manager.registerWorker("default", "worker-1", () -> hb, () -> {}); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(RecordWorkerHeartbeatRequest.class); + verify(blockingStub, timeout(VERIFY_TIMEOUT_MS).atLeastOnce()) + .recordWorkerHeartbeat(captor.capture()); + assertFalse(captor.getValue().getWorkerHeartbeat(0).hasEnvironment()); + } + @Test public void testIntervalValidation() { HeartbeatManager hm = new HeartbeatManager(service, "test-identity", Duration.ofSeconds(30)); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java new file mode 100644 index 0000000000..1c84aa37b0 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java @@ -0,0 +1,113 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.*; + +import io.temporal.api.worker.v1.EnvironmentInfo; +import io.temporal.api.worker.v1.EnvironmentInfo.Architecture; +import io.temporal.api.worker.v1.EnvironmentInfo.HostingEnvironment; +import io.temporal.api.worker.v1.EnvironmentInfo.HostingEnvironment.HostingEnvironmentType; +import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.junit.Test; + +public class WorkerEnvironmentInfoTest { + + @Test + public void detectReportsJvmRuntimeAndPlatform() { + EnvironmentInfo info = WorkerEnvironmentInfo.detect(); + + assertEquals(1, info.getRuntimesCount()); + assertEquals(RuntimeType.RUNTIME_TYPE_JVM, info.getRuntimes(0).getType()); + assertEquals(System.getProperty("java.version"), info.getRuntimes(0).getVersion()); + + Architecture expectedArchitecture = WorkerEnvironmentInfo.detectArchitecture(); + assertTrue(info.hasPlatform()); + switch (info.getPlatform().getVariantCase()) { + case LINUX: + assertEquals(expectedArchitecture, info.getPlatform().getLinux().getArchitecture()); + assertFalse(info.getPlatform().getLinux().getVersion().isEmpty()); + break; + case MACOS: + assertEquals(expectedArchitecture, info.getPlatform().getMacos().getArchitecture()); + break; + case WINDOWS: + assertEquals(expectedArchitecture, info.getPlatform().getWindows().getArchitecture()); + break; + default: + fail("unexpected platform variant " + info.getPlatform().getVariantCase()); + } + } + + @Test + public void detectHostingEnvironments() { + Map env = new HashMap<>(); + env.put("KUBERNETES_SERVICE_HOST", "10.0.0.1"); + env.put("ECS_CONTAINER_METADATA_URI", "http://169.254.170.2/v3"); + env.put("WEBSITE_SITE_NAME", "my-site"); + env.put("WEBSITE_PLATFORM_VERSION", " 1.2.3 "); + env.put("FUNCTIONS_EXTENSION_VERSION", "~4"); + env.put("GAE_SERVICE", " "); + + // Docker is detected from the host filesystem, so exclude it to keep the test host-independent. + List environments = + WorkerEnvironmentInfo.detectHostingEnvironments(env::get).stream() + .filter(e -> e.getType() != HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_DOCKER) + .collect(Collectors.toList()); + + assertEquals( + Arrays.asList( + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_K8S, + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AWS_ECS, + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AZURE_APP_SERVICE, + HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_AZURE_FUNCTIONS), + environments.stream().map(HostingEnvironment::getType).collect(Collectors.toList())); + assertEquals("1.2.3", environments.get(2).getVersion()); + assertEquals("~4", environments.get(3).getVersion()); + + assertTrue( + WorkerEnvironmentInfo.detectHostingEnvironments(name -> null).stream() + .noneMatch(e -> e.getType() != HostingEnvironmentType.HOSTING_ENVIRONMENT_TYPE_DOCKER)); + } + + @Test + public void cgroupsIndicateDocker() { + assertFalse(WorkerEnvironmentInfo.cgroupsIndicateDocker(Collections.singletonList("0::/"))); + assertTrue( + WorkerEnvironmentInfo.cgroupsIndicateDocker( + Arrays.asList("12:pids:/docker/abc123", "0::/"))); + assertTrue( + WorkerEnvironmentInfo.cgroupsIndicateDocker( + Collections.singletonList("0::/system.slice/docker-abc123.scope"))); + assertFalse( + WorkerEnvironmentInfo.cgroupsIndicateDocker( + Collections.singletonList("0::/system.slice/docker-abc123.service"))); + assertFalse( + WorkerEnvironmentInfo.cgroupsIndicateDocker( + Collections.singletonList("0::/kubepods/besteffort/pod123/dockerish"))); + } + + @Test + public void windowsVersion() { + assertEquals("11", WorkerEnvironmentInfo.windowsVersion("Windows 11", "10.0")); + assertEquals("10.0", WorkerEnvironmentInfo.windowsVersion("Windows 10", "10.0")); + assertEquals("8.1", WorkerEnvironmentInfo.windowsVersion("Windows 8.1", "6.3")); + assertEquals("10.0", WorkerEnvironmentInfo.windowsVersion("Windows Server 2022", "10.0")); + assertEquals("5.1", WorkerEnvironmentInfo.windowsVersion("Windows XP", "5.1")); + assertEquals("10.0", WorkerEnvironmentInfo.windowsVersion("Windows NT (unknown)", "10.0")); + assertEquals("6.2", WorkerEnvironmentInfo.windowsVersion("Windows", "6.2")); + } + + @Test + public void javaMajorVersion() { + assertEquals(8, WorkerEnvironmentInfo.javaMajorVersion("1.8")); + assertEquals(11, WorkerEnvironmentInfo.javaMajorVersion("11")); + assertEquals(21, WorkerEnvironmentInfo.javaMajorVersion("21.0.1")); + assertEquals(0, WorkerEnvironmentInfo.javaMajorVersion(null)); + assertEquals(0, WorkerEnvironmentInfo.javaMajorVersion("unknown")); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java index 390efe1e7d..1f0e5d4647 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java @@ -14,6 +14,7 @@ import io.temporal.activity.ActivityMethod; import io.temporal.api.enums.v1.TaskQueueType; import io.temporal.api.enums.v1.WorkerStatus; +import io.temporal.api.worker.v1.EnvironmentInfo; import io.temporal.api.worker.v1.WorkerHeartbeat; import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; import io.temporal.api.workflowservice.v1.ShutdownWorkerRequest; @@ -77,51 +78,9 @@ public OperationHandler operation() { */ @Test public void activeTaskQueueTypesEvaluatedAtShutdownTime() throws Exception { - WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); - when(service.getServerCapabilities()) - .thenReturn(() -> GetSystemInfoResponse.Capabilities.newBuilder().build()); - WorkflowServiceGrpc.WorkflowServiceFutureStub futureStub = mock(WorkflowServiceGrpc.WorkflowServiceFutureStub.class); - when(service.futureStub()).thenReturn(futureStub); - when(futureStub.shutdownWorker(any(ShutdownWorkerRequest.class))) - .thenReturn(Futures.immediateFuture(ShutdownWorkerResponse.newBuilder().build())); - - WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = - mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); - when(service.blockingStub()).thenReturn(blockingStub); - when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); - - WorkflowClient client = mock(WorkflowClient.class); - when(client.getInternal()).thenReturn(mock(WorkflowClientInternal.class)); - when(client.getWorkflowServiceStubs()).thenReturn(service); - when(client.getOptions()) - .thenReturn( - WorkflowClientOptions.newBuilder() - .setNamespace("test-ns") - .setIdentity("test-worker") - .validateAndBuildWithDefaults()); - - Scope metricsScope = new NoopScope(); - WorkflowRunLockManager runLocks = new WorkflowRunLockManager(); - WorkflowExecutorCache cache = new WorkflowExecutorCache(10, runLocks, metricsScope); - WorkflowThreadExecutor wfThreadExecutor = mock(WorkflowThreadExecutor.class); - - Worker worker = - new Worker( - client, - "test-task-queue", - WorkerFactoryOptions.newBuilder().build(), - WorkerOptions.newBuilder().build(), - metricsScope, - runLocks, - cache, - true, - wfThreadExecutor, - Collections.emptyList(), - Collections.emptyList(), - "test-worker-group", - new NamespaceCapabilities()); + Worker worker = newWorker(futureStub); // Register types AFTER worker construction. The request built by shutdown should reflect // these registrations, proving that getActiveTaskQueueTypes() is evaluated lazily. @@ -155,4 +114,85 @@ public void activeTaskQueueTypesEvaluatedAtShutdownTime() throws Exception { "ShutdownWorkerRequest sticky task queue should be derived from worker identity", captor.getValue().getStickyTaskQueue().startsWith("test-worker:")); } + + /** + * The environment is reported in every heartbeat, including the one embedded in the shutdown + * request, until the heartbeat manager reports one as accepted by the server. + */ + @Test + public void environmentInfoReportedUntilAccepted() throws Exception { + WorkflowServiceGrpc.WorkflowServiceFutureStub futureStub = + mock(WorkflowServiceGrpc.WorkflowServiceFutureStub.class); + Worker worker = newWorker(futureStub); + worker.registerWorkflowImplementationTypes(TestWorkflowImpl.class); + EnvironmentInfo environment = + EnvironmentInfo.newBuilder() + .addRuntimes( + EnvironmentInfo.Runtime.newBuilder() + .setType(EnvironmentInfo.Runtime.RuntimeType.RUNTIME_TYPE_JVM) + .setVersion("17")) + .build(); + + Supplier heartbeatSupplier = + worker.buildHeartbeatCallback("test-worker-group", environment); + worker.setHeartbeatSupplier(heartbeatSupplier); + worker.start(); + + assertEquals(environment, heartbeatSupplier.get().getEnvironment()); + assertEquals(environment, heartbeatSupplier.get().getEnvironment()); + + worker.shutdown(new ShutdownManager(), true).get(5, TimeUnit.SECONDS); + ArgumentCaptor captor = + ArgumentCaptor.forClass(ShutdownWorkerRequest.class); + verify(futureStub).shutdownWorker(captor.capture()); + assertEquals(environment, captor.getValue().getWorkerHeartbeat().getEnvironment()); + + worker.onHeartbeatAccepted(); + assertFalse(heartbeatSupplier.get().hasEnvironment()); + } + + private static Worker newWorker(WorkflowServiceGrpc.WorkflowServiceFutureStub futureStub) { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.newBuilder().build()); + + when(service.futureStub()).thenReturn(futureStub); + when(futureStub.shutdownWorker(any(ShutdownWorkerRequest.class))) + .thenReturn(Futures.immediateFuture(ShutdownWorkerResponse.newBuilder().build())); + + WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = + mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + + WorkflowClient client = mock(WorkflowClient.class); + when(client.getInternal()).thenReturn(mock(WorkflowClientInternal.class)); + when(client.getWorkflowServiceStubs()).thenReturn(service); + when(client.getOptions()) + .thenReturn( + WorkflowClientOptions.newBuilder() + .setNamespace("test-ns") + .setIdentity("test-worker") + .validateAndBuildWithDefaults()); + + Scope metricsScope = new NoopScope(); + WorkflowRunLockManager runLocks = new WorkflowRunLockManager(); + WorkflowExecutorCache cache = new WorkflowExecutorCache(10, runLocks, metricsScope); + WorkflowThreadExecutor wfThreadExecutor = mock(WorkflowThreadExecutor.class); + + return new Worker( + client, + "test-task-queue", + WorkerFactoryOptions.newBuilder().build(), + WorkerOptions.newBuilder().build(), + metricsScope, + runLocks, + cache, + true, + wfThreadExecutor, + Collections.emptyList(), + Collections.emptyList(), + "test-worker-group", + new NamespaceCapabilities()); + } }