feat(sdk): Use ConnectRPC in the go client (#2200)

### Proposed Changes

* use connect rpc clients in the go sdk
* introduced sdkconnect wrappers and interfaces to keep the same grpc
function signatures (except for grpc call options) -- these are
generated via sdk/internal/gencode
* adds a type ConnectRpcConnection which is composed of the http client,
endpoint, and connect options
* adds options to support override of new connection types and connect
options
* service updates to work over ipc with the memhttp listener
* still supports grpc gateway

### Checklist

- [ ] I have added or updated unit tests
- [ ] I have added or updated integration tests (if appropriate)
- [ ] I have added or updated documentation

### Testing Instructions

---------

Co-authored-by: Dave Mihalcik <[email protected]>

Author: elizabethhealy

.gitattributes

@@ -1,4 +1,5 @@
 docs/grpc/** linguist-generated=true
 docs/openapi/** linguist-generated=true
+sdk/sdkconnect/** linguist-generated=true
 service/policy/db/*.sql.go linguist-generated=true
-service/policy/db/models.go linguist-generated=true
+service/policy/db/models.go linguist-generated=true

.github/workflows/checks.yaml

@@ -390,6 +390,8 @@ jobs:
       - run: cd service && go get github.com/pseudomuto/protoc-gen-doc/cmd/protoc-gen-doc
       - run: cd service && go install github.com/pseudomuto/protoc-gen-doc/cmd/protoc-gen-doc
       - run: make proto-generate
+      - name: generate connect wrappers
+        run: make connect-wrapper-generate
       - name: Restore go.mod after installing protoc-gen-doc
         run: git restore {service,protocol/go}/go.{mod,sum}
       - name: validate go mod tidy
@@ -399,7 +401,7 @@ jobs:
           git restore go.sum
       - run: git diff
       - run: git diff-files --ignore-submodules
-      - name: Check that make proto-generate has run before PR submission; see above for error details
+      - name: Check that make proto-generate and connect-wrapper-generate have run before PR submission; see above for error details
         run: git diff-files --quiet --ignore-submodules
 
   ci:

Makefile

@@ -1,7 +1,7 @@
 # make
 # To run all lint checks: `LINT_OPTIONS= make lint`
 
-.PHONY: all build clean docker-build fix fmt go-lint license lint proto-generate proto-lint sdk/sdk test tidy toolcheck
+.PHONY: all build clean connect-wrapper-generate docker-build fix fmt go-lint license lint proto-generate proto-lint sdk/sdk test tidy toolcheck
 
 MODS=protocol/go lib/ocrypto lib/fixtures lib/flattening lib/identifier sdk service examples
 HAND_MODS=lib/ocrypto lib/fixtures lib/flattening lib/identifier sdk service examples
@@ -71,6 +71,9 @@ proto-generate:
 	buf generate buf.build/grpc-ecosystem/grpc-gateway -o tmp-gen --template buf.gen.grpc.docs.yaml
 	buf generate buf.build/grpc-ecosystem/grpc-gateway -o tmp-gen --template buf.gen.openapi.docs.yaml
 
+connect-wrapper-generate:
+	go run ./sdk/internal/codegen
+
 policy-sql-gen:
 	@which sqlc > /dev/null || { echo "sqlc not found, please install it: https://docs.sqlc.dev/en/stable/overview/install.html"; exit 1; }
 	sqlc generate -f service/policy/db/sqlc.yaml
@@ -95,7 +98,7 @@ clean:
 	for m in $(MODS); do (cd $$m && go clean) || exit 1; done
 	rm -f opentdf examples/examples
 
-build: proto-generate opentdf sdk/sdk examples/examples
+build: proto-generate connect-wrapper-generate opentdf sdk/sdk examples/examples
 
 opentdf: $(shell find service)
 	go build -o opentdf -v service/main.go

examples/cmd/benchmark.go

@@ -106,7 +106,11 @@ func runBenchmark(cmd *cobra.Command, _ []string) error {
 			return err
 		}
 		nanoTDFConfig.EnableECDSAPolicyBinding()
-		err = nanoTDFConfig.SetKasURL(fmt.Sprintf("http://%s/kas", "localhost:8080"))
+		if insecurePlaintextConn || strings.HasPrefix(platformEndpoint, "http://") {
+			err = nanoTDFConfig.SetKasURL(fmt.Sprintf("http://%s/kas", "localhost:8080"))
+		} else {
+			err = nanoTDFConfig.SetKasURL(fmt.Sprintf("https://%s/kas", "localhost:8080"))
+		}
 		if err != nil {
 			return err
 		}
@@ -123,15 +127,26 @@ func runBenchmark(cmd *cobra.Command, _ []string) error {
 		// 	}
 		// }
 	} else {
-		tdf, err := client.CreateTDF(
-			out, in,
-			sdk.WithDataAttributes(dataAttributes...),
-			sdk.WithKasInformation(
+		opts := []sdk.TDFOption{sdk.WithDataAttributes(dataAttributes...), sdk.WithAutoconfigure(false)}
+		if insecurePlaintextConn || strings.HasPrefix(platformEndpoint, "http://") {
+			opts = append(opts, sdk.WithKasInformation(
 				sdk.KASInfo{
-					URL:       "http://" + "localhost:8080",
+					URL:       "http://localhost:8080",
 					PublicKey: "",
 				}),
-			sdk.WithAutoconfigure(false))
+			)
+		} else {
+			opts = append(opts, sdk.WithKasInformation(
+				sdk.KASInfo{
+					URL:       "https://localhost:8080",
+					PublicKey: "",
+				}),
+			)
+		}
+		tdf, err := client.CreateTDF(
+			out, in,
+			opts...,
+		)
 		if err != nil {
 			return err
 		}

examples/cmd/benchmark_bulk.go

@@ -61,7 +61,12 @@ func runBenchmarkBulk(cmd *cobra.Command, _ []string) error {
 			return err
 		}
 		nanoTDFConfig.EnableECDSAPolicyBinding()
-		err = nanoTDFConfig.SetKasURL(fmt.Sprintf("http://%s/kas", "localhost:8080"))
+		// if plaintext or platform endpoint is http, set kas url to http, otherwise https
+		if insecurePlaintextConn || strings.HasPrefix(platformEndpoint, "http://") {
+			err = nanoTDFConfig.SetKasURL(fmt.Sprintf("http://%s/kas", "localhost:8080"))
+		} else {
+			err = nanoTDFConfig.SetKasURL(fmt.Sprintf("https://%s/kas", "localhost:8080"))
+		}
 		if err != nil {
 			return err
 		}
@@ -78,15 +83,26 @@ func runBenchmarkBulk(cmd *cobra.Command, _ []string) error {
 			}
 		}
 	} else {
-		tdf, err := client.CreateTDF(
-			out, in,
-			sdk.WithDataAttributes(dataAttributes...),
-			sdk.WithKasInformation(
+		opts := []sdk.TDFOption{sdk.WithDataAttributes(dataAttributes...), sdk.WithAutoconfigure(false)}
+		if insecurePlaintextConn || strings.HasPrefix(platformEndpoint, "http://") {
+			opts = append(opts, sdk.WithKasInformation(
 				sdk.KASInfo{
-					URL:       "http://" + "localhost:8080",
+					URL:       "http://localhost:8080",
 					PublicKey: "",
 				}),
-			sdk.WithAutoconfigure(false))
+			)
+		} else {
+			opts = append(opts, sdk.WithKasInformation(
+				sdk.KASInfo{
+					URL:       "https://localhost:8080",
+					PublicKey: "",
+				}),
+			)
+		}
+		tdf, err := client.CreateTDF(
+			out, in,
+			opts...,
+		)
 		if err != nil {
 			return err
 		}

examples/cmd/examples.go

@@ -28,7 +28,7 @@ func init() {
 	log.SetFlags(log.LstdFlags | log.Llongfile)
 	f := ExamplesCmd.PersistentFlags()
 	f.StringVarP(&clientCredentials, "creds", "", "opentdf-sdk:secret", "client id:secret credentials")
-	f.StringVarP(&platformEndpoint, "platformEndpoint", "e", "http://localhost:8080", "Platform Endpoint")
+	f.StringVarP(&platformEndpoint, "platformEndpoint", "e", "https://localhost:8080", "Platform Endpoint")
 	f.StringVarP(&tokenEndpoint, "tokenEndpoint", "t", "http://localhost:8888/auth/realms/opentdf/protocol/openid-connect/token", "OAuth token endpoint")
 	f.BoolVar(&storeCollectionHeaders, "storeCollectionHeaders", false, "Store collection headers")
 	f.BoolVar(&insecurePlaintextConn, "insecurePlaintextConn", false, "Use insecure plaintext connection")
@@ -39,6 +39,7 @@ func newSDK() (*sdk.SDK, error) {
 	resolver.SetDefaultScheme("passthrough")
 	opts := []sdk.Option{}
 	if insecurePlaintextConn {
+		platformEndpoint = strings.Replace(platformEndpoint, "https://", "http://", 1)
 		opts = append(opts, sdk.WithInsecurePlaintextConn())
 	}
 	if insecureSkipVerify {

sdk/audit/metadata_adding_interceptor.go

@@ -3,6 +3,7 @@ package audit
 import (
 	"context"
 
+	"connectrpc.com/connect"
 	"github.com/google/uuid"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 	"google.golang.org/grpc"
@@ -46,3 +47,33 @@ func MetadataAddingClientInterceptor(
 
 	return err
 }
+
+func MetadataAddingConnectInterceptor() connect.UnaryInterceptorFunc {
+	return connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
+		return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
+			// Only apply to outgoing client requests
+			if !req.Spec().IsClient {
+				return next(ctx, req)
+			}
+
+			// Get any existing request ID from context
+			requestID, ok := ctx.Value(RequestIDContextKey).(uuid.UUID)
+			if !ok || requestID == uuid.Nil {
+				requestID = uuid.New()
+			}
+			req.Header().Set(string(RequestIDHeaderKey), requestID.String())
+
+			// Add the request IP to a custom header so it is preserved
+			if requestIP, okIP := ctx.Value(RequestIPContextKey).(string); okIP {
+				req.Header().Set(string(RequestIPHeaderKey), requestIP)
+			}
+
+			// Add the actor ID from the request so it is preserved if we need it
+			if actorID, okAct := ctx.Value(ActorIDContextKey).(string); okAct {
+				req.Header().Set(string(ActorIDHeaderKey), actorID)
+			}
+
+			return next(ctx, req)
+		}
+	})
+}