feat(kas): collect metrics

.github/workflows/checks.yaml

@@ -157,6 +157,67 @@ jobs:
       - name: validate custom entity rego policy
         run: test/rego/custom-entity.bats
 
+  benchmark:
+    name: benchmark tests
+    runs-on: ubuntu-22.04
+    env:
+      TLS_ENABLED: "true"
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
+      - uses: actions/setup-go@cdcb36043654635271a94b9a6d1392de5bb323a7
+        with:
+          go-version-file: 'service/go.mod'
+          check-latest: false
+          cache-dependency-path: |
+            service/go.sum
+            examples/go.sum
+            protocol/go/go.sum
+            sdk/go.sum
+      - if: env.IS_RELEASE_BRANCH == 'true'
+        run: ./.github/scripts/work-init.sh
+      - run: go mod download
+      - run: go mod verify
+      - name: Install mkcert
+        run: |
+          sudo apt-get install -y libnss3-tools
+          curl -JLO "https://dl.filippo.io/mkcert/latest?for=linux/amd64"
+          chmod +x mkcert-v*-linux-amd64
+          sudo cp mkcert-v*-linux-amd64 /usr/local/bin/mkcert
+      - run: |
+          .github/scripts/init-temp-keys.sh
+          mkcert -install
+          mkcert -cert-file ./keys/platform.crt -key-file ./keys/platform-key.pem localhost
+          cp opentdf-dev.yaml opentdf.yaml
+          yq eval '.server.tls.enabled = true' -i opentdf.yaml
+          yq eval '.trace = {"enabled":true,"folder":"traces"}' -i opentdf.yaml
+      - name: Added Trusted Certs
+        run: |
+          sudo chmod -R 777 ./keys
+          sudo apt-get install -y ca-certificates
+          sudo cp ./keys/localhost.crt /usr/local/share/ca-certificates
+          sudo update-ca-certificates
+      - run: docker compose up -d --wait --wait-timeout 240 || (docker compose logs && exit 1)
+      - run: go run ./service provision keycloak
+      - run: go run ./service provision fixtures
+      - uses: JarvusInnovations/background-action@2428e7b970a846423095c79d43f759abf979a635
+        name: start server in background
+        with:
+          run: >
+            go build -o opentdf -v service/main.go
+            && .github/scripts/watch.sh opentdf.yaml ./opentdf start
+          wait-on: |
+            tcp:localhost:8080
+          log-output-if: true
+          wait-for: 90s
+      - name: build examples
+        run: cd examples && go build -o examples .
+      - name: run tdf3 benchmark tests
+        run: ./examples/examples benchmark --count=5000 --concurrent=50
+      - name: run nanotdf benchmark tests
+        run: ./examples/examples benchmark --storeCollectionHeaders=false --tdf=nanotdf --count=5000 --concurrent=50
+      - name: collect the metrics from the benchmark tests
+        run: ./examples/examples metrics
+
   image:
     name: image build
     runs-on: ubuntu-22.04
@@ -232,6 +293,7 @@ jobs:
       - go
       - image
       - integration
+      - benchmark
       - license
       - platform-xtest
       - otdfctl-test

.gitignore

@@ -45,3 +45,4 @@ sensitive.txt.tdf
 keys/
 /examples/sensitive.txt.ntdf
 sensitive.txt.ntdf
+traces/

examples/cmd/benchmark.go

@@ -0,0 +1,228 @@
+package cmd
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/opentdf/platform/sdk"
+	"github.com/spf13/cobra"
+)
+
+type TDFFormat string
+
+const (
+	TDF3    TDFFormat = "tdf3"
+	NanoTDF TDFFormat = "nanotdf"
+)
+
+func (f *TDFFormat) String() string {
+	return string(*f)
+}
+
+func (f *TDFFormat) Set(value string) error {
+	switch value {
+	case "tdf3", "nanotdf":
+		*f = TDFFormat(value)
+		return nil
+	default:
+		return errors.New("invalid TDF format")
+	}
+}
+
+func (f *TDFFormat) Type() string {
+	return "TDFFormat"
+}
+
+type BenchmarkConfig struct {
+	TDFFormat          TDFFormat
+	ConcurrentRequests int
+	RequestCount       int
+	RequestsPerSecond  int
+	TimeoutSeconds     int
+}
+
+var config BenchmarkConfig
+
+func init() {
+	benchmarkCmd := &cobra.Command{
+		Use:   "benchmark",
+		Short: "OpenTDF benchmark tool",
+		Long:  `A OpenTDF benchmark tool to measure throughput and latency with configurable concurrency.`,
+		RunE:  runBenchmark,
+	}
+
+	benchmarkCmd.Flags().IntVar(&config.ConcurrentRequests, "concurrent", 10, "Number of concurrent requests")
+	benchmarkCmd.Flags().IntVar(&config.RequestCount, "count", 100, "Total number of requests")
+	benchmarkCmd.Flags().IntVar(&config.RequestsPerSecond, "rps", 50, "Requests per second limit")
+	benchmarkCmd.Flags().IntVar(&config.TimeoutSeconds, "timeout", 30, "Timeout in seconds")
+	benchmarkCmd.Flags().Var(&config.TDFFormat, "tdf", "TDF format (tdf3 or nanotdf)")
+	ExamplesCmd.AddCommand(benchmarkCmd)
+}
+
+func runBenchmark(cmd *cobra.Command, args []string) error {
+	in := strings.NewReader("Hello, World!")
+
+	// Create new offline client
+	client, err := newSDK()
+	if err != nil {
+		return err
+	}
+
+	out := os.Stdout
+	if outputName != "-" {
+		out, err = os.Create("sensitive.txt.tdf")
+		if err != nil {
+			return err
+		}
+	}
+	defer func() {
+		if outputName != "-" {
+			out.Close()
+		}
+	}()
+
+	var dataAttributes = []string{"https://example.com/attr/attr1/value/value1"}
+	if config.TDFFormat == NanoTDF {
+		nanoTDFConfig, err := client.NewNanoTDFConfig()
+		if err != nil {
+			return err
+		}
+		nanoTDFConfig.SetAttributes(dataAttributes)
+		nanoTDFConfig.EnableECDSAPolicyBinding()
+		err = nanoTDFConfig.SetKasURL(fmt.Sprintf("http://%s/kas", "localhost:8080"))
+		if err != nil {
+			return err
+		}
+
+		_, err = client.CreateNanoTDF(out, in, *nanoTDFConfig)
+		if err != nil {
+			return err
+		}
+
+		if outputName != "-" {
+			err = cat(cmd, outputName)
+			if err != nil {
+				return err
+			}
+		}
+	} else {
+		tdf, err :=
+			client.CreateTDF(
+				out, in,
+				sdk.WithDataAttributes(dataAttributes...),
+				sdk.WithKasInformation(
+					sdk.KASInfo{
+						URL:       fmt.Sprintf("http://%s", "localhost:8080"),
+						PublicKey: "",
+					}),
+				sdk.WithAutoconfigure(false))
+		if err != nil {
+			return err
+		}
+
+		manifestJSON, err := json.MarshalIndent(tdf.Manifest(), "", "  ")
+		if err != nil {
+			return err
+		}
+		cmd.Println(string(manifestJSON))
+	}
+
+	var wg sync.WaitGroup
+	requests := make(chan struct{}, config.ConcurrentRequests)
+	results := make(chan time.Duration, config.RequestCount)
+	errors := make(chan error, config.RequestCount)
+
+	// Function to perform the operation
+	operation := func() {
+		defer wg.Done()
+		start := time.Now()
+
+		file, err := os.Open("sensitive.txt.tdf")
+		if err != nil {
+			errors <- fmt.Errorf("file open error: %v", err)
+			return
+		}
+		defer file.Close()
+
+		if config.TDFFormat == NanoTDF {
+			_, err = client.ReadNanoTDF(io.Discard, file)
+			if err != nil {
+				errors <- fmt.Errorf("ReadNanoTDF error: %v", err)
+				return
+			}
+		} else {
+			tdfreader, err := client.LoadTDF(file)
+			if err != nil {
+				errors <- fmt.Errorf("LoadTDF error: %v", err)
+				return
+			}
+
+			_, err = io.Copy(io.Discard, tdfreader)
+			if err != nil && err != io.EOF {
+				errors <- fmt.Errorf("read error: %v", err)
+				return
+			}
+		}
+
+		results <- time.Since(start)
+	}
+
+	// Start the benchmark
+	startTime := time.Now()
+	for i := 0; i < config.RequestCount; i++ {
+		wg.Add(1)
+		requests <- struct{}{}
+		go func() {
+			defer func() { <-requests }()
+			operation()
+		}()
+	}
+
+	wg.Wait()
+	close(results)
+	close(errors)
+
+	// Count errors and collect error messages
+	errorCount := 0
+	errorMsgs := make(map[string]int)
+	for err := range errors {
+		errorCount++
+		errorMsgs[err.Error()]++
+	}
+
+	successCount := 0
+	var totalDuration time.Duration
+	for result := range results {
+		successCount++
+		totalDuration += result
+	}
+
+	totalTime := time.Since(startTime)
+	averageLatency := totalDuration / time.Duration(successCount)
+	throughput := float64(successCount) / totalTime.Seconds()
+
+	// Print results
+	cmd.Printf("\nBenchmark Results:\n")
+	cmd.Printf("Total Requests: %d\n", config.RequestCount)
+	cmd.Printf("Successful Requests: %d\n", successCount)
+	cmd.Printf("Failed Requests: %d\n", errorCount)
+	cmd.Printf("Concurrent Requests: %d\n", config.ConcurrentRequests)
+	cmd.Printf("Total Time: %s\n", totalTime)
+	cmd.Printf("Average Latency: %s\n", averageLatency)
+	cmd.Printf("Throughput: %.2f requests/second\n", throughput)
+
+	if errorCount > 0 {
+		cmd.Printf("\nError Summary:\n")
+		for errMsg, count := range errorMsgs {
+			cmd.Printf("%s: %d occurrences\n", errMsg, count)
+		}
+	}
+
+	return nil
+}

examples/cmd/decrypt.go

@@ -4,10 +4,11 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	"github.com/spf13/cobra"
 	"io"
 	"os"
 	"path/filepath"
+
+	"github.com/spf13/cobra"
 )
 
 func init() {

examples/cmd/examples.go

@@ -13,9 +13,12 @@ import (
 )
 
 var (
-	platformEndpoint  string
-	clientCredentials string
-	tokenEndpoint     string
+	platformEndpoint       string
+	clientCredentials      string
+	tokenEndpoint          string
+	storeCollectionHeaders bool
+	insecurePlaintextConn  bool
+	insecureSkipVerify     bool
 )
 
 var ExamplesCmd = &cobra.Command{
@@ -24,14 +27,27 @@ var ExamplesCmd = &cobra.Command{
 
 func init() {
 	log.SetFlags(log.LstdFlags | log.Llongfile)
-	ExamplesCmd.PersistentFlags().StringVarP(&clientCredentials, "creds", "", "opentdf-sdk:secret", "client id:secret credentials")
-	ExamplesCmd.PersistentFlags().StringVarP(&platformEndpoint, "platformEndpoint", "e", "localhost:8080", "Platform Endpoint")
-	ExamplesCmd.PersistentFlags().StringVarP(&tokenEndpoint, "tokenEndpoint", "t", "http://localhost:8888/auth/realms/opentdf/protocol/openid-connect/token", "OAuth token endpoint")
+	f := ExamplesCmd.PersistentFlags()
+	f.StringVarP(&clientCredentials, "creds", "", "opentdf-sdk:secret", "client id:secret credentials")
+	f.StringVarP(&platformEndpoint, "platformEndpoint", "e", "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")
+	f.BoolVar(&insecureSkipVerify, "insecureSkipVerify", false, "Skip server certificate verification")
 }
 
 func newSDK() (*sdk.SDK, error) {
 	resolver.SetDefaultScheme("passthrough")
-	opts := []sdk.Option{sdk.WithStoreCollectionHeaders()}
+	opts := []sdk.Option{}
+	if insecurePlaintextConn {
+		opts = append(opts, sdk.WithInsecurePlaintextConn())
+	}
+	if insecureSkipVerify {
+		opts = append(opts, sdk.WithInsecureSkipVerifyConn())
+	}
+	if storeCollectionHeaders {
+		opts = append(opts, sdk.WithStoreCollectionHeaders())
+	}
 	if clientCredentials != "" {
 		i := strings.Index(clientCredentials, ":")
 		if i < 0 {