Merge branch 'main' into unflake-test-pls

Author: heavycrystal

docker-compose-dev.yml

@@ -8,7 +8,14 @@ x-catalog-config: &catalog-config
   PEERDB_CATALOG_DATABASE: postgres
 
 x-flow-worker-env: &flow-worker-env
+  # For Temporal Cloud, this will look like: 
+  # <yournamespace>.<id>.tmprl.cloud:7233
   TEMPORAL_HOST_PORT: temporal:7233
+  PEERDB_TEMPORAL_NAMESPACE: default
+  # For the below 2 cert and key variables,
+  # use yml multiline syntax with '|'
+  TEMPORAL_CLIENT_CERT:
+  TEMPORAL_CLIENT_KEY:
   # For GCS, these will be your HMAC keys instead
   # For more information:
   # https://cloud.google.com/storage/docs/authentication/managing-hmackeys
@@ -109,8 +116,7 @@ services:
       - 8112:8112
       - 8113:8113
     environment:
-      <<: [*catalog-config]
-      TEMPORAL_HOST_PORT: temporal:7233
+      <<: [*catalog-config, *flow-worker-env]
     depends_on:
       temporal-admin-tools:
         condition: service_healthy
@@ -122,7 +128,7 @@ services:
       dockerfile: stacks/flow.Dockerfile
       target: flow-snapshot-worker
     environment:
-      TEMPORAL_HOST_PORT: temporal:7233
+      <<: [*flow-worker-env]
     depends_on:
       temporal-admin-tools:
         condition: service_healthy

docker-compose.yml

@@ -9,6 +9,9 @@ x-catalog-config: &catalog-config
 
 x-flow-worker-env: &flow-worker-env
   TEMPORAL_HOST_PORT: temporal:7233
+  TEMPORAL_CLIENT_CERT:
+  TEMPORAL_CLIENT_KEY:
+  PEERDB_TEMPORAL_NAMESPACE: default
   # For GCS, these will be your HMAC keys instead
   # For more information:
   # https://cloud.google.com/storage/docs/authentication/managing-hmackeys
@@ -21,9 +24,6 @@ x-flow-worker-env: &flow-worker-env
   # enables exporting of mirror metrics to Prometheus for visualization using Grafana
   ENABLE_METRICS: "true"
 
-x-peerdb-temporal-namespace: &peerdb-temporal-namespace
-  PEERDB_TEMPORAL_NAMESPACE: default
-
 services:
   catalog:
     container_name: catalog
@@ -100,8 +100,7 @@ services:
       - 8112:8112
       - 8113:8113
     environment:
-      <<: [*catalog-config, *peerdb-temporal-namespace]
-      TEMPORAL_HOST_PORT: temporal:7233
+      <<: [*catalog-config, *flow-worker-env]
     depends_on:
       temporal-admin-tools:
         condition: service_healthy
@@ -110,8 +109,7 @@ services:
     container_name: flow-snapshot-worker
     image: ghcr.io/peerdb-io/flow-snapshot-worker:latest-dev
     environment:
-      <<: [*peerdb-temporal-namespace]
-      TEMPORAL_HOST_PORT: temporal:7233
+      <<: [*flow-worker-env]
     depends_on:
       temporal-admin-tools:
         condition: service_healthy
@@ -120,7 +118,7 @@ services:
     container_name: flow-worker1
     image: ghcr.io/peerdb-io/flow-worker:latest-dev
     environment:
-      <<: [*catalog-config, *flow-worker-env, *peerdb-temporal-namespace]
+      <<: [*catalog-config, *flow-worker-env]
       METRICS_SERVER: 0.0.0.0:6061
     ports:
       - 6060:6060
@@ -133,7 +131,7 @@ services:
     container_name: flow-worker2
     image: ghcr.io/peerdb-io/flow-worker:latest-dev
     environment:
-      <<: [*catalog-config, *flow-worker-env, *peerdb-temporal-namespace]
+      <<: [*catalog-config, *flow-worker-env]
       METRICS_SERVER: 0.0.0.0:6063
     ports:
       - 6062:6062
@@ -149,7 +147,7 @@ services:
     container_name: flow-worker3
     image: ghcr.io/peerdb-io/flow-worker:latest-dev
     environment:
-      <<: [*catalog-config, *flow-worker-env, *peerdb-temporal-namespace]
+      <<: [*catalog-config, *flow-worker-env]
       METRICS_SERVER: 0.0.0.0:6065
     ports:
       - 6064:6064

flow/cmd/api.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net"
 	"net/http"
@@ -26,6 +27,8 @@ type APIServerParams struct {
 	GatewayPort       uint
 	TemporalHostPort  string
 	TemporalNamespace string
+	TemporalCert      string
+	TemporalKey       string
 }
 
 // setupGRPCGatewayServer sets up the grpc-gateway mux
@@ -57,11 +60,23 @@ func setupGRPCGatewayServer(args *APIServerParams) (*http.Server, error) {
 
 func APIMain(args *APIServerParams) error {
 	ctx := args.ctx
-
-	tc, err := client.Dial(client.Options{
+	clientOptions := client.Options{
 		HostPort:  args.TemporalHostPort,
 		Namespace: args.TemporalNamespace,
-	})
+	}
+	if args.TemporalCert != "" && args.TemporalKey != "" {
+		cert, err := tls.X509KeyPair([]byte(args.TemporalCert), []byte(args.TemporalKey))
+		if err != nil {
+			return fmt.Errorf("unable to obtain temporal key pair: %w", err)
+		}
+
+		connOptions := client.ConnectionOptions{
+			TLS: &tls.Config{Certificates: []tls.Certificate{cert}},
+		}
+		clientOptions.ConnectionOptions = connOptions
+	}
+
+	tc, err := client.Dial(clientOptions)
 	if err != nil {
 		return fmt.Errorf("unable to create Temporal client: %w", err)
 	}

flow/cmd/main.go

@@ -30,6 +30,18 @@ func main() {
 		EnvVars: []string{"TEMPORAL_HOST_PORT"},
 	}
 
+	temporalCertFlag := cli.StringFlag{
+		Name:    "temporal-cert",
+		Value:   "", // default: no cert needed
+		EnvVars: []string{"TEMPORAL_CLIENT_CERT"},
+	}
+
+	temporalKeyFlag := cli.StringFlag{
+		Name:    "temporal-key",
+		Value:   "", // default: no key needed
+		EnvVars: []string{"TEMPORAL_CLIENT_KEY"},
+	}
+
 	profilingFlag := &cli.BoolFlag{
 		Name:    "enable-profiling",
 		Value:   false, // Default is off
@@ -79,6 +91,8 @@ func main() {
 						PyroscopeServer:   ctx.String("pyroscope-server-address"),
 						MetricsServer:     ctx.String("metrics-server"),
 						TemporalNamespace: ctx.String("temporal-namespace"),
+						TemporalCert:      ctx.String("temporal-cert"),
+						TemporalKey:       ctx.String("temporal-key"),
 					})
 				},
 				Flags: []cli.Flag{
@@ -88,6 +102,8 @@ func main() {
 					pyroscopeServerFlag,
 					metricsServerFlag,
 					temporalNamespaceFlag,
+					&temporalCertFlag,
+					&temporalKeyFlag,
 				},
 			},
 			{
@@ -97,11 +113,15 @@ func main() {
 					return SnapshotWorkerMain(&SnapshotWorkerOptions{
 						TemporalHostPort:  temporalHostPort,
 						TemporalNamespace: ctx.String("temporal-namespace"),
+						TemporalCert:      ctx.String("temporal-cert"),
+						TemporalKey:       ctx.String("temporal-key"),
 					})
 				},
 				Flags: []cli.Flag{
 					temporalHostPortFlag,
 					temporalNamespaceFlag,
+					&temporalCertFlag,
+					&temporalKeyFlag,
 				},
 			},
 			{
@@ -119,6 +139,8 @@ func main() {
 					},
 					temporalHostPortFlag,
 					temporalNamespaceFlag,
+					&temporalCertFlag,
+					&temporalKeyFlag,
 				},
 				Action: func(ctx *cli.Context) error {
 					temporalHostPort := ctx.String("temporal-host-port")
@@ -129,6 +151,8 @@ func main() {
 						TemporalHostPort:  temporalHostPort,
 						GatewayPort:       ctx.Uint("gateway-port"),
 						TemporalNamespace: ctx.String("temporal-namespace"),
+						TemporalCert:      ctx.String("temporal-cert"),
+						TemporalKey:       ctx.String("temporal-key"),
 					})
 				},
 			},

flow/cmd/snapshot_worker.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"crypto/tls"
 	"fmt"
 
 	"github.com/PeerDB-io/peer-flow/activities"
@@ -14,6 +15,8 @@ import (
 type SnapshotWorkerOptions struct {
 	TemporalHostPort  string
 	TemporalNamespace string
+	TemporalCert      string
+	TemporalKey       string
 }
 
 func SnapshotWorkerMain(opts *SnapshotWorkerOptions) error {
@@ -22,6 +25,17 @@ func SnapshotWorkerMain(opts *SnapshotWorkerOptions) error {
 		Namespace: opts.TemporalNamespace,
 	}
 
+	if opts.TemporalCert != "" && opts.TemporalKey != "" {
+		cert, err := tls.X509KeyPair([]byte(opts.TemporalCert), []byte(opts.TemporalKey))
+		if err != nil {
+			return fmt.Errorf("unable to obtain temporal key pair: %w", err)
+		}
+
+		connOptions := client.ConnectionOptions{
+			TLS: &tls.Config{Certificates: []tls.Certificate{cert}},
+		}
+		clientOptions.ConnectionOptions = connOptions
+	}
 	c, err := client.Dial(clientOptions)
 	if err != nil {
 		return fmt.Errorf("unable to create Temporal client: %w", err)

flow/cmd/worker.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"crypto/tls"
 	"fmt"
 	"os"
 	"os/signal"
@@ -31,6 +32,8 @@ type WorkerOptions struct {
 	PyroscopeServer   string
 	MetricsServer     string
 	TemporalNamespace string
+	TemporalCert      string
+	TemporalKey       string
 }
 
 func setupPyroscope(opts *WorkerOptions) {
@@ -95,6 +98,19 @@ func WorkerMain(opts *WorkerOptions) error {
 		HostPort:  opts.TemporalHostPort,
 		Namespace: opts.TemporalNamespace,
 	}
+
+	if opts.TemporalCert != "" && opts.TemporalKey != "" {
+		cert, err := tls.X509KeyPair([]byte(opts.TemporalCert), []byte(opts.TemporalKey))
+		if err != nil {
+			return fmt.Errorf("unable to obtain temporal key pair: %w", err)
+		}
+
+		connOptions := client.ConnectionOptions{
+			TLS: &tls.Config{Certificates: []tls.Certificate{cert}},
+		}
+		clientOptions.ConnectionOptions = connOptions
+	}
+
 	if opts.EnableMetrics {
 		clientOptions.MetricsHandler = sdktally.NewMetricsHandler(newPrometheusScope(
 			prometheus.Configuration{
@@ -115,6 +131,7 @@ func WorkerMain(opts *WorkerOptions) error {
 	if err != nil {
 		return fmt.Errorf("unable to create Temporal client: %w", err)
 	}
+	log.Info("Created temporal client")
 	defer c.Close()
 
 	w := worker.New(c, shared.PeerFlowTaskQueue, worker.Options{})

nexus/Cargo.lock

@@ -10,7 +10,6 @@ anyhow = "1.0"
 async-trait = "0.1"
 catalog = { path = "../catalog" }
 flow-rs = { path = "../flow-rs" }
-lazy_static = "1.4"
 pem = "3.0"
 pt = { path = "../pt" }
 sqlparser = { path = "../sqlparser-rs", features = ["visitor"] }