Migrate Beam Go to use the new Opaque Protocol Buffers

sdks/go/cmd/beamctl/cmd/artifact.go

@@ -95,7 +95,7 @@ func listFn(cmd *cobra.Command, args []string) error {
 	}
 
 	for _, a := range md.GetManifest().GetArtifact() {
-		cmd.Println(a.Name)
+		cmd.Println(a.GetName())
 	}
 	return nil
 }

sdks/go/container/boot.go

@@ -151,7 +151,7 @@ func main() {
 
 	enableGoogleCloudProfiler := strings.Contains(options, enableGoogleCloudProfilerOption)
 	if enableGoogleCloudProfiler {
-		err := configureGoogleCloudProfilerEnvVars(ctx, logger, info.Metadata)
+		err := configureGoogleCloudProfilerEnvVars(ctx, logger, info.GetMetadata())
 		if err != nil {
 			logger.Printf(ctx, "could not configure Google Cloud Profiler variables, got %v", err)
 		}

sdks/go/container/boot_test.go

@@ -29,11 +29,11 @@ import (
 )
 
 func TestEnsureEndpointsSet_AllSet(t *testing.T) {
-	provisionInfo := &fnpb.ProvisionInfo{
-		LoggingEndpoint:  &pipepb.ApiServiceDescriptor{Url: "testLoggingEndpointUrl"},
-		ArtifactEndpoint: &pipepb.ApiServiceDescriptor{Url: "testArtifactEndpointUrl"},
-		ControlEndpoint:  &pipepb.ApiServiceDescriptor{Url: "testControlEndpointUrl"},
-	}
+	provisionInfo := fnpb.ProvisionInfo_builder{
+		LoggingEndpoint:  pipepb.ApiServiceDescriptor_builder{Url: "testLoggingEndpointUrl"}.Build(),
+		ArtifactEndpoint: pipepb.ApiServiceDescriptor_builder{Url: "testArtifactEndpointUrl"}.Build(),
+		ControlEndpoint:  pipepb.ApiServiceDescriptor_builder{Url: "testControlEndpointUrl"}.Build(),
+	}.Build()
 	*loggingEndpoint = ""
 	*artifactEndpoint = ""
 	*controlEndpoint = ""
@@ -53,11 +53,11 @@ func TestEnsureEndpointsSet_AllSet(t *testing.T) {
 }
 
 func TestEnsureEndpointsSet_OneMissing(t *testing.T) {
-	provisionInfo := &fnpb.ProvisionInfo{
-		LoggingEndpoint:  &pipepb.ApiServiceDescriptor{Url: "testLoggingEndpointUrl"},
-		ArtifactEndpoint: &pipepb.ApiServiceDescriptor{Url: "testArtifactEndpointUrl"},
-		ControlEndpoint:  &pipepb.ApiServiceDescriptor{Url: ""},
-	}
+	provisionInfo := fnpb.ProvisionInfo_builder{
+		LoggingEndpoint:  pipepb.ApiServiceDescriptor_builder{Url: "testLoggingEndpointUrl"}.Build(),
+		ArtifactEndpoint: pipepb.ApiServiceDescriptor_builder{Url: "testArtifactEndpointUrl"}.Build(),
+		ControlEndpoint:  pipepb.ApiServiceDescriptor_builder{Url: ""}.Build(),
+	}.Build()
 	*loggingEndpoint = ""
 	*artifactEndpoint = ""
 	*controlEndpoint = ""
@@ -85,7 +85,7 @@ func TestGetGoWorkerArtifactName_NoArtifacts(t *testing.T) {
 
 func TestGetGoWorkerArtifactName_OneArtifact(t *testing.T) {
 	artifact := constructArtifactInformation(t, artifact.URNGoWorkerBinaryRole, "test/path", "sha")
-	artifacts := []*pipepb.ArtifactInformation{&artifact}
+	artifacts := []*pipepb.ArtifactInformation{artifact}
 
 	val, err := getGoWorkerArtifactName(context.Background(), &tools.Logger{}, artifacts)
 	if err != nil {
@@ -99,7 +99,7 @@ func TestGetGoWorkerArtifactName_OneArtifact(t *testing.T) {
 func TestGetGoWorkerArtifactName_MultipleArtifactsFirstIsWorker(t *testing.T) {
 	artifact1 := constructArtifactInformation(t, artifact.URNGoWorkerBinaryRole, "test/path", "sha")
 	artifact2 := constructArtifactInformation(t, "other role", "test/path2", "sha")
-	artifacts := []*pipepb.ArtifactInformation{&artifact1, &artifact2}
+	artifacts := []*pipepb.ArtifactInformation{artifact1, artifact2}
 
 	val, err := getGoWorkerArtifactName(context.Background(), &tools.Logger{}, artifacts)
 	if err != nil {
@@ -113,7 +113,7 @@ func TestGetGoWorkerArtifactName_MultipleArtifactsFirstIsWorker(t *testing.T) {
 func TestGetGoWorkerArtifactName_MultipleArtifactsSecondIsWorker(t *testing.T) {
 	artifact1 := constructArtifactInformation(t, "other role", "test/path", "sha")
 	artifact2 := constructArtifactInformation(t, artifact.URNGoWorkerBinaryRole, "test/path2", "sha")
-	artifacts := []*pipepb.ArtifactInformation{&artifact1, &artifact2}
+	artifacts := []*pipepb.ArtifactInformation{artifact1, artifact2}
 
 	val, err := getGoWorkerArtifactName(context.Background(), &tools.Logger{}, artifacts)
 	if err != nil {
@@ -127,7 +127,7 @@ func TestGetGoWorkerArtifactName_MultipleArtifactsSecondIsWorker(t *testing.T) {
 func TestGetGoWorkerArtifactName_MultipleArtifactsLegacyWay(t *testing.T) {
 	artifact1 := constructArtifactInformation(t, "other role", "test/path", "sha")
 	artifact2 := constructArtifactInformation(t, "other role", "worker", "sha")
-	artifacts := []*pipepb.ArtifactInformation{&artifact1, &artifact2}
+	artifacts := []*pipepb.ArtifactInformation{artifact1, artifact2}
 
 	val, err := getGoWorkerArtifactName(context.Background(), &tools.Logger{}, artifacts)
 	if err != nil {
@@ -141,7 +141,7 @@ func TestGetGoWorkerArtifactName_MultipleArtifactsLegacyWay(t *testing.T) {
 func TestGetGoWorkerArtifactName_MultipleArtifactsNoneMatch(t *testing.T) {
 	artifact1 := constructArtifactInformation(t, "other role", "test/path", "sha")
 	artifact2 := constructArtifactInformation(t, "other role", "test/path2", "sha")
-	artifacts := []*pipepb.ArtifactInformation{&artifact1, &artifact2}
+	artifacts := []*pipepb.ArtifactInformation{artifact1, artifact2}
 
 	_, err := getGoWorkerArtifactName(context.Background(), &tools.Logger{}, artifacts)
 	if err == nil {
@@ -193,16 +193,16 @@ func TestCopyExe(t *testing.T) {
 	}
 }
 
-func constructArtifactInformation(t *testing.T, roleUrn string, path string, sha string) pipepb.ArtifactInformation {
+func constructArtifactInformation(t *testing.T, roleUrn string, path string, sha string) *pipepb.ArtifactInformation {
 	t.Helper()
 
-	typePayload, _ := proto.Marshal(&pipepb.ArtifactFilePayload{Path: path, Sha256: sha})
+	typePayload, _ := proto.Marshal(pipepb.ArtifactFilePayload_builder{Path: path, Sha256: sha}.Build())
 
-	return pipepb.ArtifactInformation{
+	return pipepb.ArtifactInformation_builder{
 		RoleUrn:     roleUrn,
 		TypeUrn:     artifact.URNFileArtifact,
 		TypePayload: typePayload,
-	}
+	}.Build()
 }
 
 func TestConfigureGoogleCloudProfilerEnvVars(t *testing.T) {

sdks/go/container/tools/buffered_logging_test.go

@@ -63,11 +63,11 @@ func TestBufferedLogger(t *testing.T) {
 
 		received := catcher.msgs[0].GetLogEntries()[0]
 
-		if got, want := received.Message, "test message"; got != want {
+		if got, want := received.GetMessage(), "test message"; got != want {
 			t.Errorf("got message %q, want %q", got, want)
 		}
 
-		if got, want := received.Severity, fnpb.LogEntry_Severity_DEBUG; got != want {
+		if got, want := received.GetSeverity(), fnpb.LogEntry_Severity_DEBUG; got != want {
 			t.Errorf("got severity %v, want %v", got, want)
 		}
 	})
@@ -96,11 +96,11 @@ func TestBufferedLogger(t *testing.T) {
 		received := catcher.msgs[0].GetLogEntries()
 
 		for i, message := range received {
-			if got, want := message.Message, messages[i]; got != want {
+			if got, want := message.GetMessage(), messages[i]; got != want {
 				t.Errorf("got message %q, want %q", got, want)
 			}
 
-			if got, want := message.Severity, fnpb.LogEntry_Severity_DEBUG; got != want {
+			if got, want := message.GetSeverity(), fnpb.LogEntry_Severity_DEBUG; got != want {
 				t.Errorf("got severity %v, want %v", got, want)
 			}
 		}
@@ -125,11 +125,11 @@ func TestBufferedLogger(t *testing.T) {
 
 		received := catcher.msgs[0].GetLogEntries()[0]
 
-		if got, want := received.Message, "test error"; got != want {
+		if got, want := received.GetMessage(), "test error"; got != want {
 			t.Errorf("got message %q, want %q", got, want)
 		}
 
-		if got, want := received.Severity, fnpb.LogEntry_Severity_ERROR; got != want {
+		if got, want := received.GetSeverity(), fnpb.LogEntry_Severity_ERROR; got != want {
 			t.Errorf("got severity %v, want %v", got, want)
 		}
 	})
@@ -158,11 +158,11 @@ func TestBufferedLogger(t *testing.T) {
 		received := catcher.msgs[0].GetLogEntries()
 
 		for i, message := range received {
-			if got, want := message.Message, messages[i]; got != want {
+			if got, want := message.GetMessage(), messages[i]; got != want {
 				t.Errorf("got message %q, want %q", got, want)
 			}
 
-			if got, want := message.Severity, fnpb.LogEntry_Severity_ERROR; got != want {
+			if got, want := message.GetSeverity(), fnpb.LogEntry_Severity_ERROR; got != want {
 				t.Errorf("got severity %v, want %v", got, want)
 			}
 		}
@@ -177,11 +177,11 @@ func TestBufferedLogger(t *testing.T) {
 
 		received := catcher.msgs[0].GetLogEntries()[0]
 
-		if got, want := received.Message, "foo bar"; got != want {
+		if got, want := received.GetMessage(), "foo bar"; got != want {
 			t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got message %q, want %q", got, want)
 		}
 
-		if got, want := received.Severity, fnpb.LogEntry_Severity_DEBUG; got != want {
+		if got, want := received.GetSeverity(), fnpb.LogEntry_Severity_DEBUG; got != want {
 			t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got severity %v, want %v", got, want)
 		}
 	})
@@ -229,11 +229,11 @@ func TestBufferedLogger(t *testing.T) {
 		messages = append(messages, lastMessage)
 
 		for i, message := range received {
-			if got, want := message.Message, messages[i]; got != want {
+			if got, want := message.GetMessage(), messages[i]; got != want {
 				t.Errorf("got message %q, want %q", got, want)
 			}
 
-			if got, want := message.Severity, fnpb.LogEntry_Severity_DEBUG; got != want {
+			if got, want := message.GetSeverity(), fnpb.LogEntry_Severity_DEBUG; got != want {
 				t.Errorf("got severity %v, want %v", got, want)
 			}
 		}

sdks/go/container/tools/logging.go

@@ -86,15 +86,15 @@ func (l *Logger) Log(ctx context.Context, sev fnpb.LogEntry_Severity_Enum, messa
 		}
 	}
 
-	err := l.client.Send(&fnpb.LogEntry_List{
+	err := l.client.Send(fnpb.LogEntry_List_builder{
 		LogEntries: []*fnpb.LogEntry{
-			{
+			fnpb.LogEntry_builder{
 				Severity:  sev,
 				Timestamp: timestamppb.Now(),
 				Message:   message,
-			},
+			}.Build(),
 		},
-	})
+	}.Build())
 	if err != nil {
 		exitErr = err
 		return

sdks/go/container/tools/logging_test.go

@@ -50,11 +50,11 @@ func TestLogger(t *testing.T) {
 
 		received := catcher.msgs[0].GetLogEntries()[0]
 
-		if got, want := received.Message, "foo bar"; got != want {
+		if got, want := received.GetMessage(), "foo bar"; got != want {
 			t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got message %q, want %q", got, want)
 		}
 
-		if got, want := received.Severity, fnpb.LogEntry_Severity_DEBUG; got != want {
+		if got, want := received.GetSeverity(), fnpb.LogEntry_Severity_DEBUG; got != want {
 			t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got severity %v, want %v", got, want)
 		}
 	})
@@ -66,11 +66,11 @@ func TestLogger(t *testing.T) {
 
 		received := catcher.msgs[0].GetLogEntries()[0]
 
-		if got, want := received.Message, "failed to install dependency bar"; got != want {
+		if got, want := received.GetMessage(), "failed to install dependency bar"; got != want {
 			t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got message %q, want %q", got, want)
 		}
 
-		if got, want := received.Severity, fnpb.LogEntry_Severity_ERROR; got != want {
+		if got, want := received.GetSeverity(), fnpb.LogEntry_Severity_ERROR; got != want {
 			t.Errorf("l.Errorf(\"failed to install dependency %%v\", \"bar\"): got severity %v, want %v", got, want)
 		}
 	})

sdks/go/container/tools/provision_test.go

@@ -66,7 +66,7 @@ type ProvisionServiceServicer struct {
 }
 
 func (p ProvisionServiceServicer) GetProvisionInfo(ctx context.Context, req *fnpb.GetProvisionInfoRequest) (*fnpb.GetProvisionInfoResponse, error) {
-	return &fnpb.GetProvisionInfoResponse{Info: &fnpb.ProvisionInfo{RetrievalToken: "token"}}, nil
+	return fnpb.GetProvisionInfoResponse_builder{Info: fnpb.ProvisionInfo_builder{RetrievalToken: "token"}.Build()}.Build(), nil
 }
 
 func setup(addr *string, wg *sync.WaitGroup) {
@@ -103,7 +103,7 @@ func TestProvisionInfo(t *testing.T) {
 	if err != nil {
 		t.Errorf("error in response: %v", err)
 	}
-	want := &fnpb.ProvisionInfo{RetrievalToken: "token"}
+	want := fnpb.ProvisionInfo_builder{RetrievalToken: "token"}.Build()
 	if got.GetRetrievalToken() != want.GetRetrievalToken() {
 		t.Errorf("provision.Info() = %v, want %v", got, want)
 	}

sdks/go/examples/snippets/06schemas_test.go

@@ -31,63 +31,57 @@ import (
 )
 
 func atomicSchemaField(name string, typ pipepb.AtomicType) *pipepb.Field {
-	return &pipepb.Field{
+	return pipepb.Field_builder{
 		Name: name,
-		Type: &pipepb.FieldType{
-			TypeInfo: &pipepb.FieldType_AtomicType{
-				AtomicType: typ,
-			},
-		},
-	}
+		Type: pipepb.FieldType_builder{
+			AtomicType: typ.Enum(),
+		}.Build(),
+	}.Build()
 }
 
 func rowSchemaField(name string, typ *pipepb.Schema) *pipepb.Field {
-	return &pipepb.Field{
+	return pipepb.Field_builder{
 		Name: name,
-		Type: &pipepb.FieldType{
-			TypeInfo: &pipepb.FieldType_RowType{
-				RowType: &pipepb.RowType{
-					Schema: typ,
-				},
-			},
-		},
-	}
+		Type: pipepb.FieldType_builder{
+			RowType: pipepb.RowType_builder{
+				Schema: typ,
+			}.Build(),
+		}.Build(),
+	}.Build()
 }
 
 func listSchemaField(name string, typ *pipepb.Field) *pipepb.Field {
-	return &pipepb.Field{
+	return pipepb.Field_builder{
 		Name: name,
-		Type: &pipepb.FieldType{
-			TypeInfo: &pipepb.FieldType_ArrayType{
-				ArrayType: &pipepb.ArrayType{
-					ElementType: typ.GetType(),
-				},
-			},
-		},
-	}
+		Type: pipepb.FieldType_builder{
+			ArrayType: pipepb.ArrayType_builder{
+				ElementType: typ.GetType(),
+			}.Build(),
+		}.Build(),
+	}.Build()
 }
 
 func nillable(f *pipepb.Field) *pipepb.Field {
-	f.Type.Nullable = true
+	f.GetType().SetNullable(true)
 	return f
 }
 
 func TestSchemaTypes(t *testing.T) {
-	transactionSchema := &pipepb.Schema{
+	transactionSchema := pipepb.Schema_builder{
 		Fields: []*pipepb.Field{
 			atomicSchemaField("bank", pipepb.AtomicType_STRING),
 			atomicSchemaField("purchaseAmount", pipepb.AtomicType_DOUBLE),
 		},
-	}
-	shippingAddressSchema := &pipepb.Schema{
+	}.Build()
+	shippingAddressSchema := pipepb.Schema_builder{
 		Fields: []*pipepb.Field{
 			atomicSchemaField("streetAddress", pipepb.AtomicType_STRING),
 			atomicSchemaField("city", pipepb.AtomicType_STRING),
 			nillable(atomicSchemaField("state", pipepb.AtomicType_STRING)),
 			atomicSchemaField("country", pipepb.AtomicType_STRING),
 			atomicSchemaField("postCode", pipepb.AtomicType_STRING),
 		},
-	}
+	}.Build()
 
 	tests := []struct {
 		rt     reflect.Type
@@ -101,7 +95,7 @@ func TestSchemaTypes(t *testing.T) {
 		st: shippingAddressSchema,
 	}, {
 		rt: reflect.TypeOf(Purchase{}),
-		st: &pipepb.Schema{
+		st: pipepb.Schema_builder{
 			Fields: []*pipepb.Field{
 				atomicSchemaField("userId", pipepb.AtomicType_STRING),
 				atomicSchemaField("itemId", pipepb.AtomicType_INT64),
@@ -110,15 +104,15 @@ func TestSchemaTypes(t *testing.T) {
 				listSchemaField("transactions",
 					rowSchemaField("n/a", transactionSchema)),
 			},
-		},
+		}.Build(),
 	}, {
 		rt: tnType,
-		st: &pipepb.Schema{
+		st: pipepb.Schema_builder{
 			Fields: []*pipepb.Field{
 				atomicSchemaField("seconds", pipepb.AtomicType_INT64),
 				atomicSchemaField("nanos", pipepb.AtomicType_INT32),
 			},
-		},
+		}.Build(),
 		preReg: func(reg *schema.Registry) {
 			reg.RegisterLogicalType(schema.ToLogicalType(tnType.Name(), tnType, tnStorageType))
 		},