Merge pull request #235 from GoogleCloudPlatform/234-override-service…

…-account-in-pub-sub-trigger

fix: extract and merge overrides

internal/apiclient/iam.go

@@ -390,6 +390,12 @@ func SetCloudSpannerIAMPermission(project string, memberName string) (err error)
 	return setProjectIAMPermission(project, memberName, role)
 }
 
+// SetIntegrationInvokerPermission
+func SetIntegrationInvokerPermission(project string, memberName string) (err error) {
+	const role = "roles/integrations.integrationInvoker"
+	return setProjectIAMPermission(project, memberName, role)
+}
+
 func getNameAndProject(iamFullName string) (projectid string, name string, err error) {
 	riam := regexp.MustCompile(`^[a-zA-Z0-9-]{6,30}$`)
 

internal/client/integrations/integrations.go

@@ -262,7 +262,7 @@ type integrationConnection struct {
 
 // CreateVersion
 func CreateVersion(name string, content []byte, overridesContent []byte, snapshot string,
-	userlabel string,
+	userlabel string, grantPermission bool,
 ) (respBody []byte, err error) {
 	iversion := integrationVersion{}
 	if err = json.Unmarshal(content, &iversion); err != nil {
@@ -289,7 +289,7 @@ func CreateVersion(name string, content []byte, overridesContent []byte, snapsho
 		if err = json.Unmarshal(overridesContent, &o); err != nil {
 			return nil, err
 		}
-		if eversion, err = mergeOverrides(eversion, o); err != nil {
+		if eversion, err = mergeOverrides(eversion, o, grantPermission); err != nil {
 			return nil, err
 		}
 	}
@@ -1213,7 +1213,7 @@ func uploadAsync(name string, filePath string) error {
 		return err
 	}
 
-	if _, err := CreateVersion(name, content, nil, "", ""); err != nil {
+	if _, err := CreateVersion(name, content, nil, "", "", false); err != nil {
 		return err
 	}
 

internal/client/integrations/overrides.go

@@ -50,6 +50,7 @@ type triggeroverrides struct {
 	ProjectId                    *string           `json:"projectId,omitempty"`
 	TopicName                    *string           `json:"topicName,omitempty"`
 	APIPath                      *string           `json:"apiPath,omitempty"`
+	ServiceAccount               *string           `json:"serviceAccount,omitempty"`
 	Properties                   map[string]string `json:"properties,omitempty"`
 	CloudSchedulerServiceAccount *string           `json:"cloudSchedulerServiceAccount,omitempty"`
 	CloudSchedulerLocation       *string           `json:"cloudSchedulerLocation,omitempty"`
@@ -88,7 +89,7 @@ const (
 const authConfigValue = "{  \"@type\": \"type.googleapis.com/enterprise.crm.eventbus.authconfig.AuthConfigTaskParam\",\"authConfigId\": \""
 
 // mergeOverrides
-func mergeOverrides(eversion integrationVersionExternal, o overrides) (integrationVersionExternal, error) {
+func mergeOverrides(eversion integrationVersionExternal, o overrides, grantPermission bool) (integrationVersionExternal, error) {
 	// apply trigger overrides
 	for _, triggerOverride := range o.TriggerOverrides {
 		foundOverride := false
@@ -101,6 +102,20 @@ func mergeOverrides(eversion integrationVersionExternal, o overrides) (integrati
 					}
 					trigger.TriggerId = pubsubTrigger + *triggerOverride.ProjectId + "_" + *triggerOverride.TopicName
 					trigger.Properties["Subscription name"] = *triggerOverride.ProjectId + "_" + *triggerOverride.TopicName
+					trigger.Properties["IP Project name"] = *triggerOverride.ProjectId
+					if triggerOverride.ServiceAccount != nil {
+						serviceAccountName := fmt.Sprintf("%s@%s.iam.gserviceaccount.com", *triggerOverride.ServiceAccount, *triggerOverride.ProjectId)
+						trigger.Properties["Service account"] = serviceAccountName
+						if grantPermission {
+							// create the SA if it doesn't exist
+							if err := apiclient.CreateServiceAccount(serviceAccountName); err != nil {
+								return eversion, err
+							}
+							if err := apiclient.SetIntegrationInvokerPermission(*triggerOverride.ProjectId, serviceAccountName); err != nil {
+								clilog.Warning.Printf("Unable to update permissions for the service account: %v\n", err)
+							}
+						}
+					}
 				case "API":
 					if triggerOverride.APIPath == nil {
 						return eversion, fmt.Errorf("the field apiPath is missing from the API Trigger in overrides")
@@ -297,9 +312,15 @@ func extractOverrides(iversion integrationVersion) (overrides, error) {
 			triggerOverride := triggeroverrides{}
 			triggerOverride.ProjectId = new(string)
 			triggerOverride.TopicName = new(string)
+			triggerOverride.ServiceAccount = new(string)
 			*triggerOverride.ProjectId = strings.Split(subscription, "_")[0]
 			*triggerOverride.TopicName = strings.Split(subscription, "_")[1]
 			triggerOverride.TriggerNumber = triggerConfig.TriggerNumber
+			if sa, err := apiclient.GetComputeEngineDefaultServiceAccount(apiclient.GetProjectID()); err != nil {
+				if sa != triggerConfig.Properties["Service account"] {
+					*triggerOverride.ServiceAccount = strings.Split(triggerConfig.Properties["Service account"], "@")[0]
+				}
+			}
 			taskOverrides.TriggerOverrides = append(taskOverrides.TriggerOverrides, triggerOverride)
 		case "CLOUD_SCHEDULER":
 			triggerOverride := triggeroverrides{}

internal/cmd/integrations/apply.go

@@ -134,7 +134,8 @@ var ApplyCmd = &cobra.Command{
 			return err
 		}
 
-		if err = processIntegration(overridesFile, integrationFolder, configVarsFolder, pipeline); err != nil {
+		if err = processIntegration(overridesFile, integrationFolder,
+			configVarsFolder, pipeline, grantPermission); err != nil {
 			return err
 		}
 
@@ -160,9 +161,9 @@ func init() {
 	ApplyCmd.Flags().StringVarP(&userLabel, "userlabel", "u",
 		"", "Integration version userlabel")
 	ApplyCmd.Flags().StringVarP(&serviceAccountName, "sa", "",
-		"", "Service Account name for the connection")
+		"", "Service Account name for the connection or integration trigger")
 	ApplyCmd.Flags().StringVarP(&serviceAccountProject, "sp", "",
-		"", "Service Account Project for the connection. Default is the connection's project id")
+		"", "Service Account Project for the connection or integraton trigger.")
 	ApplyCmd.Flags().StringVarP(&encryptionKey, "encryption-keyid", "k",
 		"", "Cloud KMS key for decrypting Auth Config; Format = locations/*/keyRings/*/cryptoKeys/*")
 	ApplyCmd.Flags().StringVarP(&env, "env", "e",
@@ -523,7 +524,8 @@ func processSfdcChannels(sfdcchannelsFolder string) (err error) {
 	return nil
 }
 
-func processIntegration(overridesFile string, integrationFolder string, configVarsFolder string, pipeline string) (err error) {
+func processIntegration(overridesFile string, integrationFolder string,
+	configVarsFolder string, pipeline string, grantPermission bool) (err error) {
 	rJSONFiles := regexp.MustCompile(`(\S*)\.json`)
 
 	var integrationNames []string
@@ -563,7 +565,7 @@ func processIntegration(overridesFile string, integrationFolder string, configVa
 		}
 		clilog.Info.Printf("Create integration %s\n", getFilenameWithoutExtension(integrationNames[0]))
 		respBody, err := integrations.CreateVersion(getFilenameWithoutExtension(integrationNames[0]),
-			integrationBytes, overridesBytes, "", userLabel)
+			integrationBytes, overridesBytes, "", userLabel, grantPermission)
 		if err != nil {
 			return err
 		}

internal/cmd/integrations/create.go

@@ -62,12 +62,13 @@ var CreateCmd = &cobra.Command{
 			}
 		}
 
-		_, err = integrations.CreateVersion(name, content, overridesContent, snapshot, userLabel)
+		_, err = integrations.CreateVersion(name, content, overridesContent, snapshot, userLabel, grantPermission)
 		return err
 	},
 }
 
 var integrationFile, overridesFile string
+var grantPermission bool
 
 func init() {
 	var name string
@@ -82,6 +83,8 @@ func init() {
 		"", "Integration version snapshot number")
 	CreateCmd.Flags().StringVarP(&userLabel, "userlabel", "u",
 		"", "Integration version userlabel")
+	CreateCmd.Flags().BoolVarP(&grantPermission, "grant-permission", "g",
+		false, "Grant the service account permission for integration triggers; default is false")
 
 	_ = CreateCmd.MarkFlagRequired("name")
 	_ = CreateCmd.MarkFlagRequired("file")

test/pubsub_overrides.json

@@ -1,20 +1,25 @@
 {
-    "trigger_overrides": [{
-        "triggerNumber": "1",
-        "triggerType": "CLOUD_PUBSUB_EXTERNAL",
-        "projectId": "my-project",
-        "topicName": "topic"
-    }],
-    "task_overrides": [{
-        "taskId": "1",
-        "task": "GenericRestV2Task",
-        "parameters":  {
-            "url": {
-                "key": "url",
-                "value": {
-                    "stringValue": "https://httpbin.org/ip"
+    "trigger_overrides": [
+        {
+            "triggerNumber": "1",
+            "triggerType": "CLOUD_PUBSUB_EXTERNAL",
+            "projectId": "my-project",
+            "topicName": "topic",
+            "serviceAccount": "my-sa"
+        }
+    ],
+    "task_overrides": [
+        {
+            "taskId": "1",
+            "task": "GenericRestV2Task",
+            "parameters": {
+                "url": {
+                    "key": "url",
+                    "value": {
+                        "stringValue": "https://httpbin.org/ip"
+                    }
                 }
             }
         }
-    }]
-}
+    ]
+}