GODRIVER-2810 Switch to polling monitoring when running within a FaaS environment (mongodb#1376)

Author: prestonvasquez

Dockerfile

@@ -31,6 +31,8 @@ RUN export DEBIAN_FRONTEND=noninteractive && \
     software-properties-common \
     gpg \
     apt-utils \
+    libc6-dev \
+    gcc \
     make && \
   sudo update-ca-certificates && \
   rm -rf /var/lib/apt/lists/*

internal/driverutil/hello.go

@@ -0,0 +1,128 @@
+// Copyright (C) MongoDB, Inc. 2023-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package driverutil
+
+import (
+	"os"
+	"strings"
+)
+
+const AwsLambdaPrefix = "AWS_Lambda_"
+
+const (
+	// FaaS environment variable names
+
+	// EnvVarAWSExecutionEnv is the AWS Execution environment variable.
+	EnvVarAWSExecutionEnv = "AWS_EXECUTION_ENV"
+	// EnvVarAWSLambdaRuntimeAPI is the AWS Lambda runtime API variable.
+	EnvVarAWSLambdaRuntimeAPI = "AWS_LAMBDA_RUNTIME_API"
+	// EnvVarFunctionsWorkerRuntime is the functions worker runtime variable.
+	EnvVarFunctionsWorkerRuntime = "FUNCTIONS_WORKER_RUNTIME"
+	// EnvVarKService is the K Service variable.
+	EnvVarKService = "K_SERVICE"
+	// EnvVarFunctionName is the function name variable.
+	EnvVarFunctionName = "FUNCTION_NAME"
+	// EnvVarVercel is the Vercel variable.
+	EnvVarVercel = "VERCEL"
+	// EnvVarK8s is the K8s veriable.
+	EnvVarK8s = "KUBERNETES_SERVICE_HOST"
+)
+
+const (
+	// FaaS environment variable names
+
+	// EnvVarAWSRegion is the AWS region variable.
+	EnvVarAWSRegion = "AWS_REGION"
+	// EnvVarAWSLambdaFunctionMemorySize is the AWS Lambda function memory size variable.
+	EnvVarAWSLambdaFunctionMemorySize = "AWS_LAMBDA_FUNCTION_MEMORY_SIZE"
+	// EnvVarFunctionMemoryMB is the function memory in megabytes variable.
+	EnvVarFunctionMemoryMB = "FUNCTION_MEMORY_MB"
+	// EnvVarFunctionTimeoutSec is the function timeout in seconds variable.
+	EnvVarFunctionTimeoutSec = "FUNCTION_TIMEOUT_SEC"
+	// EnvVarFunctionRegion is the function region variable.
+	EnvVarFunctionRegion = "FUNCTION_REGION"
+	// EnvVarVercelRegion is the Vercel region variable.
+	EnvVarVercelRegion = "VERCEL_REGION"
+)
+
+const (
+	// FaaS environment names used by the client
+
+	// EnvNameAWSLambda is the AWS Lambda environment name.
+	EnvNameAWSLambda = "aws.lambda"
+	// EnvNameAzureFunc is the Azure Function environment name.
+	EnvNameAzureFunc = "azure.func"
+	// EnvNameGCPFunc is the Google Cloud Function environment name.
+	EnvNameGCPFunc = "gcp.func"
+	// EnvNameVercel is the Vercel environment name.
+	EnvNameVercel = "vercel"
+)
+
+// GetFaasEnvName parses the FaaS environment variable name and returns the
+// corresponding name used by the client. If none of the variables or variables
+// for multiple names are populated the client.env value MUST be entirely
+// omitted. When variables for multiple "client.env.name" values are present,
+// "vercel" takes precedence over "aws.lambda"; any other combination MUST cause
+// "client.env" to be entirely omitted.
+func GetFaasEnvName() string {
+	envVars := []string{
+		EnvVarAWSExecutionEnv,
+		EnvVarAWSLambdaRuntimeAPI,
+		EnvVarFunctionsWorkerRuntime,
+		EnvVarKService,
+		EnvVarFunctionName,
+		EnvVarVercel,
+	}
+
+	// If none of the variables are populated the client.env value MUST be
+	// entirely omitted.
+	names := make(map[string]struct{})
+
+	for _, envVar := range envVars {
+		val := os.Getenv(envVar)
+		if val == "" {
+			continue
+		}
+
+		var name string
+
+		switch envVar {
+		case EnvVarAWSExecutionEnv:
+			if !strings.HasPrefix(val, AwsLambdaPrefix) {
+				continue
+			}
+
+			name = EnvNameAWSLambda
+		case EnvVarAWSLambdaRuntimeAPI:
+			name = EnvNameAWSLambda
+		case EnvVarFunctionsWorkerRuntime:
+			name = EnvNameAzureFunc
+		case EnvVarKService, EnvVarFunctionName:
+			name = EnvNameGCPFunc
+		case EnvVarVercel:
+			// "vercel" takes precedence over "aws.lambda".
+			delete(names, EnvNameAWSLambda)
+
+			name = EnvNameVercel
+		}
+
+		names[name] = struct{}{}
+		if len(names) > 1 {
+			// If multiple names are populated the client.env value
+			// MUST be entirely omitted.
+			names = nil
+
+			break
+		}
+	}
+
+	for name := range names {
+		return name
+	}
+
+	return ""
+}

internal/driverutil/const.go renamed to internal/driverutil/operation.go

@@ -27,11 +27,12 @@ const timeout = 60 * time.Second
 // event durations, as well as the number of heartbeats, commands, and open
 // connections.
 type eventListener struct {
-	commandCount      int
-	commandDuration   int64
-	heartbeatCount    int
-	heartbeatDuration int64
-	openConnections   int
+	commandCount          int
+	commandDuration       int64
+	heartbeatAwaitedCount int
+	heartbeatCount        int
+	heartbeatDuration     int64
+	openConnections       int
 }
 
 // commandMonitor initializes an event.CommandMonitor that will count the number
@@ -61,11 +62,19 @@ func (listener *eventListener) serverMonitor() *event.ServerMonitor {
 	succeeded := func(e *event.ServerHeartbeatSucceededEvent) {
 		listener.heartbeatCount++
 		listener.heartbeatDuration += e.DurationNanos
+
+		if e.Awaited {
+			listener.heartbeatAwaitedCount++
+		}
 	}
 
 	failed := func(e *event.ServerHeartbeatFailedEvent) {
 		listener.heartbeatCount++
 		listener.heartbeatDuration += e.DurationNanos
+
+		if e.Awaited {
+			listener.heartbeatAwaitedCount++
+		}
 	}
 
 	return &event.ServerMonitor{
@@ -150,6 +159,12 @@ func handler(ctx context.Context, request events.APIGatewayProxyRequest) (events
 		return gateway500(), fmt.Errorf("failed to delete: %w", err)
 	}
 
+	// Driver must switch to polling monitoring when running within a FaaS
+	// environment.
+	if listener.heartbeatAwaitedCount > 0 {
+		return gateway500(), fmt.Errorf("FaaS environment fialed to switch to polling")
+	}
+
 	var avgCmdDur float64
 	if count := listener.commandCount; count != 0 {
 		avgCmdDur = float64(listener.commandDuration) / float64(count)

internal/test/faas/awslambda/mongodb/main.go

@@ -55,31 +55,18 @@ func TestHandshakeProse(t *testing.T) {
 		return elems
 	}
 
-	const (
-		envVarAWSExecutionEnv             = "AWS_EXECUTION_ENV"
-		envVarAWSRegion                   = "AWS_REGION"
-		envVarAWSLambdaFunctionMemorySize = "AWS_LAMBDA_FUNCTION_MEMORY_SIZE"
-		envVarFunctionsWorkerRuntime      = "FUNCTIONS_WORKER_RUNTIME"
-		envVarKService                    = "K_SERVICE"
-		envVarFunctionMemoryMB            = "FUNCTION_MEMORY_MB"
-		envVarFunctionTimeoutSec          = "FUNCTION_TIMEOUT_SEC"
-		envVarFunctionRegion              = "FUNCTION_REGION"
-		envVarVercel                      = "VERCEL"
-		envVarVercelRegion                = "VERCEL_REGION"
-	)
-
 	// Reset the environment variables to avoid environment namespace
 	// collision.
-	t.Setenv(envVarAWSExecutionEnv, "")
-	t.Setenv(envVarFunctionsWorkerRuntime, "")
-	t.Setenv(envVarKService, "")
-	t.Setenv(envVarVercel, "")
-	t.Setenv(envVarAWSRegion, "")
-	t.Setenv(envVarAWSLambdaFunctionMemorySize, "")
-	t.Setenv(envVarFunctionMemoryMB, "")
-	t.Setenv(envVarFunctionTimeoutSec, "")
-	t.Setenv(envVarFunctionRegion, "")
-	t.Setenv(envVarVercelRegion, "")
+	t.Setenv("AWS_EXECUTION_ENV", "")
+	t.Setenv("FUNCTIONS_WORKER_RUNTIME", "")
+	t.Setenv("K_SERVICE", "")
+	t.Setenv("VERCEL", "")
+	t.Setenv("AWS_REGION", "")
+	t.Setenv("AWS_LAMBDA_FUNCTION_MEMORY_SIZE", "")
+	t.Setenv("FUNCTION_MEMORY_MB", "")
+	t.Setenv("FUNCTION_TIMEOUT_SEC", "")
+	t.Setenv("FUNCTION_REGION", "")
+	t.Setenv("VERCEL_REGION", "")
 
 	for _, test := range []struct {
 		name string
@@ -89,9 +76,9 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "1. valid AWS",
 			env: map[string]string{
-				envVarAWSExecutionEnv:             "AWS_Lambda_java8",
-				envVarAWSRegion:                   "us-east-2",
-				envVarAWSLambdaFunctionMemorySize: "1024",
+				"AWS_EXECUTION_ENV":               "AWS_Lambda_java8",
+				"AWS_REGION":                      "us-east-2",
+				"AWS_LAMBDA_FUNCTION_MEMORY_SIZE": "1024",
 			},
 			want: clientMetadata(bson.D{
 				{Key: "name", Value: "aws.lambda"},
@@ -102,7 +89,7 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "2. valid Azure",
 			env: map[string]string{
-				envVarFunctionsWorkerRuntime: "node",
+				"FUNCTIONS_WORKER_RUNTIME": "node",
 			},
 			want: clientMetadata(bson.D{
 				{Key: "name", Value: "azure.func"},
@@ -111,10 +98,10 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "3. valid GCP",
 			env: map[string]string{
-				envVarKService:           "servicename",
-				envVarFunctionMemoryMB:   "1024",
-				envVarFunctionTimeoutSec: "60",
-				envVarFunctionRegion:     "us-central1",
+				"K_SERVICE":            "servicename",
+				"FUNCTION_MEMORY_MB":   "1024",
+				"FUNCTION_TIMEOUT_SEC": "60",
+				"FUNCTION_REGION":      "us-central1",
 			},
 			want: clientMetadata(bson.D{
 				{Key: "name", Value: "gcp.func"},
@@ -126,8 +113,8 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "4. valid Vercel",
 			env: map[string]string{
-				envVarVercel:       "1",
-				envVarVercelRegion: "cdg1",
+				"VERCEL":        "1",
+				"VERCEL_REGION": "cdg1",
 			},
 			want: clientMetadata(bson.D{
 				{Key: "name", Value: "vercel"},
@@ -137,16 +124,16 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "5. invalid multiple providers",
 			env: map[string]string{
-				envVarAWSExecutionEnv:        "AWS_Lambda_java8",
-				envVarFunctionsWorkerRuntime: "node",
+				"AWS_EXECUTION_ENV":        "AWS_Lambda_java8",
+				"FUNCTIONS_WORKER_RUNTIME": "node",
 			},
 			want: clientMetadata(nil),
 		},
 		{
 			name: "6. invalid long string",
 			env: map[string]string{
-				envVarAWSExecutionEnv: "AWS_Lambda_java8",
-				envVarAWSRegion: func() string {
+				"AWS_EXECUTION_ENV": "AWS_Lambda_java8",
+				"AWS_REGION": func() string {
 					var s string
 					for i := 0; i < 512; i++ {
 						s += "a"
@@ -161,8 +148,8 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "7. invalid wrong types",
 			env: map[string]string{
-				envVarAWSExecutionEnv:             "AWS_Lambda_java8",
-				envVarAWSLambdaFunctionMemorySize: "big",
+				"AWS_EXECUTION_ENV":               "AWS_Lambda_java8",
+				"AWS_LAMBDA_FUNCTION_MEMORY_SIZE": "big",
 			},
 			want: clientMetadata(bson.D{
 				{Key: "name", Value: "aws.lambda"},
@@ -171,7 +158,7 @@ func TestHandshakeProse(t *testing.T) {
 		{
 			name: "8. Invalid - AWS_EXECUTION_ENV does not start with \"AWS_Lambda_\"",
 			env: map[string]string{
-				envVarAWSExecutionEnv: "EC2",
+				"AWS_EXECUTION_ENV": "EC2",
 			},
 			want: clientMetadata(nil),
 		},
@@ -188,32 +175,27 @@ func TestHandshakeProse(t *testing.T) {
 			require.NoError(mt, err, "Ping error: %v", err)
 
 			messages := mt.GetProxiedMessages()
+			handshakeMessage := messages[:1][0]
 
-			// First two messages are handshake messages
-			for idx, pair := range messages[:2] {
-				hello := handshake.LegacyHello
-				//  Expect "hello" command name with API version.
-				if os.Getenv("REQUIRE_API_VERSION") == "true" {
-					hello = "hello"
-				}
-
-				assert.Equal(mt, pair.CommandName, hello, "expected and actual command name at index %d are different", idx)
+			hello := handshake.LegacyHello
+			if os.Getenv("REQUIRE_API_VERSION") == "true" {
+				hello = "hello"
+			}
 
-				sent := pair.Sent
+			assert.Equal(mt, hello, handshakeMessage.CommandName)
 
-				// Lookup the "client" field in the command document.
-				clientVal, err := sent.Command.LookupErr("client")
-				require.NoError(mt, err, "expected command %s at index %d to contain client field", sent.Command, idx)
+			// Lookup the "client" field in the command document.
+			clientVal, err := handshakeMessage.Sent.Command.LookupErr("client")
+			require.NoError(mt, err, "expected command %s to contain client field", handshakeMessage.Sent.Command)
 
-				got, ok := clientVal.DocumentOK()
-				require.True(mt, ok, "expected client field to be a document, got %s", clientVal.Type)
+			got, ok := clientVal.DocumentOK()
+			require.True(mt, ok, "expected client field to be a document, got %s", clientVal.Type)
 
-				wantBytes, err := bson.Marshal(test.want)
-				require.NoError(mt, err, "error marshaling want document: %v", err)
+			wantBytes, err := bson.Marshal(test.want)
+			require.NoError(mt, err, "error marshaling want document: %v", err)
 
-				want := bsoncore.Document(wantBytes)
-				assert.Equal(mt, want, got, "want: %v, got: %v", want, got)
-			}
+			want := bsoncore.Document(wantBytes)
+			assert.Equal(mt, want, got, "want: %v, got: %v", want, got)
 		})
 	}
 }

mongo/integration/handshake_test.go

@@ -32,7 +32,8 @@ func TestSDAMProse(t *testing.T) {
 	heartbeatIntervalMtOpts := mtest.NewOptions().
 		ClientOptions(heartbeatIntervalClientOpts).
 		CreateCollection(false).
-		ClientType(mtest.Proxy)
+		ClientType(mtest.Proxy).
+		MinServerVersion("4.4") // RTT Monitor / Streaming protocol is not supported for versions < 4.4.
 	mt.RunOpts("heartbeats processed more frequently", heartbeatIntervalMtOpts, func(mt *mtest.T) {
 		// Test that setting heartbeat interval to 500ms causes the client to process heartbeats
 		// approximately every 500ms instead of the default 10s. Note that a Client doesn't

mongo/integration/sdam_prose_test.go

@@ -599,6 +599,8 @@ func setClientOptionsFromURIOptions(clientOpts *options.ClientOptions, uriOpts b
 			clientOpts.SetTimeout(time.Duration(value.(int32)) * time.Millisecond)
 		case "serverselectiontimeoutms":
 			clientOpts.SetServerSelectionTimeout(time.Duration(value.(int32)) * time.Millisecond)
+		case "servermonitoringmode":
+			clientOpts.SetServerMonitoringMode(value.(string))
 		default:
 			return fmt.Errorf("unrecognized URI option %s", key)
 		}