Cadence® Connection Examples Using TLS

Cadence
Technical

December 02, 2022
By Rony Rahman

In Cadence® 0.24.0, TLS support was introduced for the gRPC transport (https://github.com/uber/cadence/pull/4606). In this blog we provide a couple of sample code snippets—one using the Go SDK and the other using the Java SDK—to demonstrate how to establish connection to the Cadence server using TLS. For both the snippets, we assume a X.509 Certificate Authority (CA) certificate from the Cadence server in PEM-encoded (ASCII) format called ca-certificate.pem.

Go

To create a Worker using Go, the following code snippet can be used:

package main
import (
	"fmt"
	"github.com/uber-go/tally"
	apiv1 "github.com/uber/cadence-idl/go/proto/api/v1"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	"go.uber.org/cadence/compatibility"
	"go.uber.org/cadence/worker"
	"go.uber.org/yarpc"
	"go.uber.org/yarpc/transport/grpc"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"crypto/tls"
	"crypto/x509"
	"io/ioutil"
	"go.uber.org/yarpc/peer"
	"go.uber.org/yarpc/peer/hostport"
	"google.golang.org/grpc/credentials"
)

var HostPort = "<HOST>:<PORT>"
var Domain = "<DOMAIN>"
var TaskListName = "<TASK>"
var ClientName = "<CLIENT>"
var CadenceService = "cadence-frontend"

func main() {
     …………………
     …………………
     …………………
     worker := getWorker(buildLogger(), buildCadenceClient())
     …………………
     …………………
     …………………
}

func buildLogger() *zap.Logger {
     config := zap.NewDevelopmentConfig()
     config.Level.SetLevel(zapcore.InfoLevel)

     var err error
     logger, err := config.Build()
     if err != nil {
          panic("Failed to setup logger")
     }

     return logger
}

func buildCadenceClient() workflowserviceclient.Interface {
     grpcTransport := grpc.NewTransport()
     var dialOptions []grpc.DialOption

     caCert, err := ioutil.ReadFile("ca-certificate.pem")
     if err != nil {
          fmt.Printf("Failed to load server CA certificate: %v", zap.Error(err))
     }

     caCertPool := x509.NewCertPool()
     if !caCertPool.AppendCertsFromPEM(caCert) {
          fmt.Errorf("Failed to add server CA's certificate")
     }

     tlsConfig := tls.Config{
          RootCAs: caCertPool,
     }

     creds := credentials.NewTLS(&tlsConfig)
     dialOptions = append(dialOptions, grpc.DialerCredentials(creds))

     dialer := grpcTransport.NewDialer(dialOptions...)
     outbound := grpcTransport.NewOutbound(
                        peer.NewSingle(hostport.PeerIdentifier(HostPort), dialer)
                 )

     dispatcher := yarpc.NewDispatcher(yarpc.Config{
          Name: ClientName,
          Outbounds: yarpc.Outbounds{
               CadenceService: {Unary: outbound},
          },
     })
     if err := dispatcher.Start(); err != nil {
          panic("Failed to start dispatcher")
     }

     clientConfig := dispatcher.ClientConfig(CadenceService)

     return compatibility.NewThrift2ProtoAdapter(
          apiv1.NewDomainAPIYARPCClient(clientConfig),
          apiv1.NewWorkflowAPIYARPCClient(clientConfig),
          apiv1.NewWorkerAPIYARPCClient(clientConfig),
          apiv1.NewVisibilityAPIYARPCClient(clientConfig),
     )
}

func getWorker(logger *zap.Logger, service workflowserviceclient.Interface) worker.Worker {
     workerOptions := worker.Options{
          Logger:       logger,
          MetricsScope: tally.NewTestScope(TaskListName, map[string]string{}),
     }

     worker := worker.New(
          service,
          Domain,
          TaskListName,
          workerOptions)

     return worker
}

100

101

102

103

104

105

106

107

108

109

package main

import (

"fmt"

"github.com/uber-go/tally"

apiv1 "github.com/uber/cadence-idl/go/proto/api/v1"

"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"

"go.uber.org/cadence/compatibility"

"go.uber.org/cadence/worker"

"go.uber.org/yarpc"

"go.uber.org/yarpc/transport/grpc"

"go.uber.org/zap"

"go.uber.org/zap/zapcore"

"crypto/tls"

"crypto/x509"

"io/ioutil"

"go.uber.org/yarpc/peer"

"go.uber.org/yarpc/peer/hostport"

"google.golang.org/grpc/credentials"

)

var HostPort = "<HOST>:<PORT>"

var Domain = "<DOMAIN>"

var TaskListName = "<TASK>"

var ClientName = "<CLIENT>"

var CadenceService = "cadence-frontend"

func main() {

…………………

worker := getWorker(buildLogger(), buildCadenceClient())

…………………

}

func buildLogger() *zap.Logger {

config := zap.NewDevelopmentConfig()

config.Level.SetLevel(zapcore.InfoLevel)

var err error

logger, err := config.Build()

if err != nil {

panic("Failed to setup logger")

}

return logger

}

func buildCadenceClient() workflowserviceclient.Interface {

grpcTransport := grpc.NewTransport()

var dialOptions []grpc.DialOption

caCert, err := ioutil.ReadFile("ca-certificate.pem")

if err != nil {

fmt.Printf("Failed to load server CA certificate: %v", zap.Error(err))

}

caCertPool := x509.NewCertPool()

if !caCertPool.AppendCertsFromPEM(caCert) {

fmt.Errorf("Failed to add server CA's certificate")

}

tlsConfig := tls.Config{

RootCAs: caCertPool,

}

creds := credentials.NewTLS(&tlsConfig)

dialOptions = append(dialOptions, grpc.DialerCredentials(creds))

dialer := grpcTransport.NewDialer(dialOptions...)

outbound := grpcTransport.NewOutbound(

peer.NewSingle(hostport.PeerIdentifier(HostPort), dialer)

)

dispatcher := yarpc.NewDispatcher(yarpc.Config{

Name: ClientName,

Outbounds: yarpc.Outbounds{

CadenceService: {Unary: outbound},

})

if err := dispatcher.Start(); err != nil {

panic("Failed to start dispatcher")

}

clientConfig := dispatcher.ClientConfig(CadenceService)

return compatibility.NewThrift2ProtoAdapter(

apiv1.NewDomainAPIYARPCClient(clientConfig),

apiv1.NewWorkflowAPIYARPCClient(clientConfig),

apiv1.NewWorkerAPIYARPCClient(clientConfig),

apiv1.NewVisibilityAPIYARPCClient(clientConfig),

)

}

func getWorker(logger *zap.Logger, service workflowserviceclient.Interface) worker.Worker {

workerOptions := worker.Options{

Logger: logger,

MetricsScope: tally.NewTestScope(TaskListName, map[string]string{}),

}

worker := worker.New(

service,

Domain,

TaskListName,

workerOptions)

return worker

}

Java

In order to use the Java SDK to establish a connection to the Cadence server over TLS, the following Maven dependencies are needed:

<dependency>
    <groupId>io.grpc</groupId>
    <artifactId>grpc-netty</artifactId>
    <version>1.28.0</version>
</dependency>
<dependency>
    <groupId>com.uber.cadence</groupId>
    <artifactId>cadence-client</artifactId>
    <version>3.7.2</version>
</dependency>
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.17.Final</version>
</dependency>

<artifactId>grpc-netty</artifactId>

</dependency>

<groupId>com.uber.cadence</groupId>

<artifactId>cadence-client</artifactId>

</dependency>

<groupId>io.netty</groupId>

<artifactId>netty-all</artifactId>

<version>4.1.17.Final</version>

</dependency>

The following code snippet demonstrates how to create a new Worker.

import com.uber.cadence.client.WorkflowClient;
import com.uber.cadence.client.WorkflowClientOptions;
import com.uber.cadence.internal.compatibility.Thrift2ProtoAdapter;
import com.uber.cadence.internal.compatibility.proto.serviceclient.IGrpcServiceStubs;
import com.uber.cadence.serviceclient.ClientOptions;
import com.uber.cadence.worker.Worker;
import com.uber.cadence.worker.WorkerFactory;
import io.grpc.netty.GrpcSslContexts;
import io.grpc.netty.NettyChannelBuilder;
import java.io.File;

……………………
……………………
……………………

     WorkflowClient workflowClient =
          WorkflowClient.newInstance(
               new Thrift2ProtoAdapter(
                    IGrpcServiceStubs.newInstance(
                         ClientOptions.newBuilder()
                              .setGRPCChannel(
                                   NettyChannelBuilder.forAddress("<HOST>", <PORT>)
                                        .useTransportSecurity()
                                        .defaultLoadBalancingPolicy("round_robin")
                                        .sslContext(GrpcSslContexts.forClient()
                                             .trustManager(new File("ca-certificate.pem"))
                                             .build()
                                        )
                                        .build()
                              )
                              .build()
                    )
               ),
               WorkflowClientOptions.newBuilder()
                    .setDomain("<DOMAIN>")
                    .build()
          );

     WorkerFactory factory = WorkerFactory.newInstance(workflowClient);
     Worker worker = factory.newWorker("<ACTIVITY>");

……………………
……………………
……………………

import com.uber.cadence.client.WorkflowClient;

import com.uber.cadence.client.WorkflowClientOptions;

import com.uber.cadence.internal.compatibility.Thrift2ProtoAdapter;

import com.uber.cadence.internal.compatibility.proto.serviceclient.IGrpcServiceStubs;

import com.uber.cadence.serviceclient.ClientOptions;

import com.uber.cadence.worker.Worker;

import com.uber.cadence.worker.WorkerFactory;

import io.grpc.netty.GrpcSslContexts;

import io.grpc.netty.NettyChannelBuilder;

import java.io.File;

……………………

WorkflowClient workflowClient =

WorkflowClient.newInstance(

new Thrift2ProtoAdapter(

IGrpcServiceStubs.newInstance(

ClientOptions.newBuilder()

.setGRPCChannel(

NettyChannelBuilder.forAddress("<HOST>", <PORT>)

.useTransportSecurity()

.defaultLoadBalancingPolicy("round_robin")

.sslContext(GrpcSslContexts.forClient()

.trustManager(new File("ca-certificate.pem"))

.build()

)

.build()

)

.build()

)

WorkflowClientOptions.newBuilder()

.setDomain("<DOMAIN>")

.build()

);

WorkerFactory factory = WorkerFactory.newInstance(workflowClient);

Worker worker = factory.newWorker("<ACTIVITY>");

……………………

Go

Java

Instaclustr Is Now Available on the Azure Marketplace

Security Advisory CVE-2024-3094 xz-utils

IoT Overdrive Part 2: Where Can We Improve?