External Storage Integration: WorkflowWorker, replay, history #3017

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

cconstable wants to merge 17 commits into main from extstore/workflow-worker

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -22,8 +22,10 @@ @@
     import io.temporal.internal.client.external.GenericWorkflowClientImpl;
     import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
     import io.temporal.internal.common.PluginUtils;
+    import io.temporal.internal.payload.storage.ExternalStorageRunner;
     import io.temporal.internal.sync.StubMarker;
     import io.temporal.internal.worker.HeartbeatManager;
+    import io.temporal.payload.storage.ExternalStorage;
     import io.temporal.serviceclient.MetricsTag;
     import io.temporal.serviceclient.WorkflowServiceStubs;
     import io.temporal.serviceclient.WorkflowServiceStubsPlugin;
@@ Expand Down Expand Up @@
       private final WorkerFactoryRegistry workerFactoryRegistry = new WorkerFactoryRegistry();
       private final String workerGroupingKey = java.util.UUID.randomUUID().toString();
       private final @Nullable HeartbeatManager heartbeatManager;
+      private final @Nullable ExternalStorageRunner externalStorage;
       /**
        * Creates client that connects to an instance of the Temporal Service. Cannot be used from within
@@ Expand Down Expand Up / @@ -106,6 +109,9 @@ public static WorkflowClient newInstance( @@
                 .getOptions()
                 .getMetricsScope()
                 .tagged(MetricsTag.defaultTags(options.getNamespace()));
+        ExternalStorage externalStorageConfig = options.getExternalStorage();
+        this.externalStorage =
+            externalStorageConfig == null ? null : ExternalStorageRunner.create(externalStorageConfig);
         this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
         this.interceptors = options.getInterceptors();
         this.workflowClientCallsInvoker = initializeClientInvoker();
@@ Expand Down Expand Up / @@ -815,6 +821,12 @@ public HeartbeatManager getHeartbeatManager() { @@
         return heartbeatManager;
       }
+      @Override
+      @Nullable
+      public ExternalStorageRunner getExternalStorage() {
+        return externalStorage;
+      }
       @Override
       public NexusStartWorkflowResponse startNexus(
           NexusStartWorkflowRequest request, Functions.Proc workflow) {
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -7,12 +7,14 @@ @@
     import io.temporal.common.converter.DataConverter;
     import io.temporal.common.converter.GlobalDataConverter;
     import io.temporal.common.interceptors.WorkflowClientInterceptor;
+    import io.temporal.payload.storage.ExternalStorage;
     import java.lang.management.ManagementFactory;
     import java.time.Duration;
     import java.util.Arrays;
     import java.util.Collections;
     import java.util.List;
     import java.util.Objects;
+    import javax.annotation.Nullable;
     /** Options for WorkflowClient configuration. */
     public final class WorkflowClientOptions {
@@ Expand Down Expand Up / @@ -52,6 +54,7 @@ public static final class Builder { @@
         private QueryRejectCondition queryRejectCondition;
         private WorkflowClientPlugin[] plugins;
         private Duration workerHeartbeatInterval;
+        private ExternalStorage externalStorage;
         private Builder() {}
@@ Expand All / @@ -68,6 +71,7 @@ private Builder(WorkflowClientOptions options) { @@
           queryRejectCondition = options.queryRejectCondition;
           plugins = options.plugins;
           workerHeartbeatInterval = options.workerHeartbeatInterval;
+          externalStorage = options.externalStorage;
         }
         public Builder setNamespace(String namespace) {
@@ Expand All @@
           return this;
         }
+        /**
+         * External storage configuration uses to store/retrieve large payloads.
+         *
+         * <p>Defaults to null.
+         */
+        @Experimental
+        public Builder setExternalStorage(@Nullable ExternalStorage externalStorage) {
+          this.externalStorage = externalStorage;
+          return this;
+        }
         /**
          * Interceptor used to intercept workflow client calls.
          *
@@ Expand Down Expand Up / @@ -180,7 +195,8 @@ public WorkflowClientOptions build() { @@
               contextPropagators,
               queryRejectCondition,
               plugins == null ? EMPTY_PLUGINS : plugins,
-              resolveHeartbeatInterval(workerHeartbeatInterval));
+              resolveHeartbeatInterval(workerHeartbeatInterval),
+              externalStorage);
         }
         /**
@@ Expand All @@
                   ? QueryRejectCondition.QUERY_REJECT_CONDITION_UNSPECIFIED
                   : queryRejectCondition,
               plugins == null ? EMPTY_PLUGINS : plugins,
-              resolveHeartbeatInterval(workerHeartbeatInterval));
+              resolveHeartbeatInterval(workerHeartbeatInterval),
+              externalStorage);
         }
         private static Duration resolveHeartbeatInterval(Duration raw) {
@@ Expand Down Expand Up @@
       private final Duration workerHeartbeatInterval;
+      private final @Nullable ExternalStorage externalStorage;
       private WorkflowClientOptions(
           String namespace,
           DataConverter dataConverter,
@@ Expand All / @@ -259,7 +278,8 @@ private WorkflowClientOptions( @@
           List<ContextPropagator> contextPropagators,
           QueryRejectCondition queryRejectCondition,
           WorkflowClientPlugin[] plugins,
-          Duration workerHeartbeatInterval) {
+          Duration workerHeartbeatInterval,
+          @Nullable ExternalStorage externalStorage) {
         this.namespace = namespace;
         this.dataConverter = dataConverter;
         this.interceptors = interceptors;
@@ Expand All / @@ -269,6 +289,7 @@ private WorkflowClientOptions( @@
         this.queryRejectCondition = queryRejectCondition;
         this.plugins = plugins;
         this.workerHeartbeatInterval = workerHeartbeatInterval;
+        this.externalStorage = externalStorage;
       }
       /**
@@ Expand All / @@ -284,6 +305,13 @@ public DataConverter getDataConverter() { @@
         return dataConverter;
       }
+      /** External storage used to offload large payloads or null when disabled. */
+      @Experimental
+      @Nullable
+      public ExternalStorage getExternalStorage() {
+        return externalStorage;
+      }
       public WorkflowClientInterceptor[] getInterceptors() {
         return interceptors;
       }
@@ Expand Down Expand Up / @@ -359,6 +387,8 @@ public String toString() { @@
             + Arrays.toString(plugins)
             + ", workerHeartbeatInterval="
             + workerHeartbeatInterval
+            + ", externalStorage="
+            + externalStorage
             + '}';
       }
@@ Expand All / @@ -376,7 +406,8 @@ public boolean equals(Object o) { @@
             && queryRejectCondition == that.queryRejectCondition
             && Arrays.equals(plugins, that.plugins)
             && com.google.common.base.Objects.equal(
-                workerHeartbeatInterval, that.workerHeartbeatInterval);
+                workerHeartbeatInterval, that.workerHeartbeatInterval)
+            && com.google.common.base.Objects.equal(externalStorage, that.externalStorage);
       }
       @Override
@@ Expand All / @@ -390,6 +421,7 @@ public int hashCode() { @@
             contextPropagators,
             queryRejectCondition,
             Arrays.hashCode(plugins),
-            workerHeartbeatInterval);
+            workerHeartbeatInterval,
+            externalStorage);
       }
     }

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

-Original file line number
+Diff line change
@@ -1,6 +1,7 @@
     package io.temporal.internal.client;
     import io.temporal.client.WorkflowClient;
+    import io.temporal.internal.payload.storage.ExternalStorageRunner;
     import io.temporal.internal.worker.HeartbeatManager;
     import io.temporal.worker.WorkerFactory;
     import io.temporal.workflow.Functions;
@@ Expand All / @@ -25,4 +26,7 @@ public interface WorkflowClientInternal { @@
       @Nullable
       HeartbeatManager getHeartbeatManager();
+      @Nullable
+      ExternalStorageRunner getExternalStorage();
     }

...-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java

-Original file line number
+Diff line change
@@ -0,0 +1,124 @@
+    package io.temporal.internal.payload.storage;
+    import io.temporal.api.common.v1.Payload;
+    import io.temporal.api.common.v1.Payloads;
+    import io.temporal.api.failure.v1.Failure;
+    import io.temporal.common.CancellationToken;
+    import io.temporal.common.converter.DataConverter;
+    import io.temporal.common.converter.DataConverterException;
+    import io.temporal.payload.context.SerializationContext;
+    import io.temporal.payload.storage.StorageDriverTargetInfo;
+    import java.lang.reflect.Type;
+    import java.util.Optional;
+    import javax.annotation.Nonnull;
+    import javax.annotation.Nullable;
+    /**
+     * A {@link DataConverter} that stores/retrieves payloads to/from external storage.
+     *
+     * <p>This This is an internal class that is not exposed to users or workflow code. The intent is to
+     * use this data converter to consolidate extstore usage within the SDK.
+     */
+    public final class ExternalStorageDataConverter implements DataConverter {
+      private final DataConverter delegate;
+      private final ExternalStorageRunner externalStorage;
+      private final @Nullable StorageDriverTargetInfo storageTarget;
+      public ExternalStorageDataConverter(
+          @Nonnull DataConverter delegate, @Nonnull ExternalStorageRunner externalStorage) {
+        this(delegate, externalStorage, null);
+      }
+      private ExternalStorageDataConverter(
+          @Nonnull DataConverter delegate,
+          @Nonnull ExternalStorageRunner externalStorage,
+          @Nullable StorageDriverTargetInfo storageTarget) {
+        this.delegate = delegate;
+        this.externalStorage = externalStorage;
+        this.storageTarget = storageTarget;
+      }
+      public ExternalStorageDataConverter withStorageTarget(
+          @Nullable StorageDriverTargetInfo storageTarget) {
+        return new ExternalStorageDataConverter(delegate, externalStorage, storageTarget);
+      }
+      @Override
+      public <T> Optional<Payload> toPayload(T value) throws DataConverterException {
+        Optional<Payload> converted = delegate.toPayload(value);
+        if (!converted.isPresent()) {
+          return converted;
+        }
+        Payloads stored = store(Payloads.newBuilder().addPayloads(converted.get()).build());
+        return Optional.of(stored.getPayloads(0));
+      }
+      @Override
+      public Optional<Payloads> toPayloads(Object... values) throws DataConverterException {
+        Optional<Payloads> converted = delegate.toPayloads(values);
+        if (!converted.isPresent()) {
+          return converted;
+        }
+        return Optional.of(store(converted.get()));
+      }
+      @Override
+      public <T> T fromPayload(Payload payload, Class<T> valueClass, Type valueType)
+          throws DataConverterException {
+        return delegate.fromPayload(retrieve(payload), valueClass, valueType);
+      }
+      @Override
+      public <T> T fromPayloads(
+          int index, Optional<Payloads> content, Class<T> parameterType, Type genericParameterType)
+          throws DataConverterException {
+        if (!content.isPresent() || index >= content.get().getPayloadsCount()) {
+          return delegate.fromPayloads(index, content, parameterType, genericParameterType);
+        }
+        Payload resolved = retrieve(content.get().getPayloads(index));
+        return delegate.fromPayload(resolved, parameterType, genericParameterType);
+      }
+      @Override
+      @Nonnull
+      public RuntimeException failureToException(@Nonnull Failure failure) {
+        return delegate.failureToException(retrieveMessage(failure));
+      }
+      @Override
+      @Nonnull
+      public Failure exceptionToFailure(@Nonnull Throwable throwable) {
+        return storeMessage(delegate.exceptionToFailure(throwable));
+      }
+      @Override
+      @Nonnull
+      public DataConverter withContext(@Nonnull SerializationContext context) {
+        return new ExternalStorageDataConverter(
+            delegate.withContext(context), externalStorage, storageTarget);
+      }
+      private Payload retrieve(Payload payload) {
+        if (!ExternalStorageReferences.isReference(payload)) {
+          return payload;
+        }
+        return retrieveMessage(Payloads.newBuilder().addPayloads(payload).build()).getPayloads(0);
+      }
+      private Payloads store(Payloads payloads) {
+        Payloads.Builder builder = payloads.toBuilder();
+        externalStorage.store(builder, storageTarget, null, CancellationToken.none());
+        return builder.build();
+      }
+      private <T extends com.google.protobuf.Message> T retrieveMessage(T message) {
+        return externalStorage.retrieve(message, CancellationToken.none());
+      }
+      private Failure storeMessage(Failure failure) {
+        Failure.Builder builder = failure.toBuilder();
+        externalStorage.store(builder, storageTarget, null, CancellationToken.none());
+        return builder.build();
+      }
+    }

...src/main/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformer.java

This file was deleted.

Navigation Menu

Uh oh!

External Storage Integration: WorkflowWorker, replay, history #3017

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!

Uh oh!

External Storage Integration: WorkflowWorker, replay, history #3017

Are you sure you want to change the base?

Uh oh!

External Storage Integration: WorkflowWorker, replay, history #3017

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!