Fix Msi support for stream analytics job

internal/services/streamanalytics/stream_analytics_job_resource.go

@@ -160,6 +160,7 @@ func resourceStreamAnalyticsJob() *pluginsdk.Resource {
 							Default:  string(streamingjobs.AuthenticationModeConnectionString),
 							ValidateFunc: validation.StringInSlice([]string{
 								string(streamingjobs.AuthenticationModeConnectionString),
+								string(streamingjobs.AuthenticationModeMsi),
 							}, false),
 						},
 
@@ -171,7 +172,7 @@ func resourceStreamAnalyticsJob() *pluginsdk.Resource {
 
 						"account_key": {
 							Type:         pluginsdk.TypeString,
-							Required:     true,
+							Optional:     true,
 							Sensitive:    true,
 							ValidateFunc: validation.StringIsNotEmpty,
 						},
@@ -279,6 +280,10 @@ func resourceStreamAnalyticsJobCreateUpdate(d *pluginsdk.ResourceData, meta inte
 	if contentStoragePolicy == string(streamingjobs.ContentStoragePolicyJobStorageAccount) {
 		if v, ok := d.GetOk("job_storage_account"); ok {
 			props.Properties.JobStorageAccount = expandJobStorageAccount(v.([]interface{}))
+
+			if *props.Properties.JobStorageAccount.AuthenticationMode == streamingjobs.AuthenticationModeMsi && d.IsNewResource() {
+				return fmt.Errorf("`authenticationMode` must be set to `ConnectionString` when creating a new stream analytics job.\nSwitch to `Msi` in an update phase once the job and system managed identity exists and is authorized to access the storage account.")
+			}
 		} else {
 			return fmt.Errorf("`job_storage_account` must be set when `content_storage_policy` is `JobStorageAccount`")
 		}
@@ -542,10 +547,17 @@ func expandJobStorageAccount(input []interface{}) *streamingjobs.JobStorageAccou
 	accountName := v["account_name"].(string)
 	accountKey := v["account_key"].(string)
 
-	return &streamingjobs.JobStorageAccount{
-		AuthenticationMode: utils.ToPtr(streamingjobs.AuthenticationMode(authenticationMode)),
-		AccountName:        utils.String(accountName),
-		AccountKey:         utils.String(accountKey),
+	if authenticationMode == string(streamingjobs.AuthenticationModeMsi) {
+		return &streamingjobs.JobStorageAccount{
+			AuthenticationMode: utils.ToPtr(streamingjobs.AuthenticationMode(authenticationMode)),
+			AccountName:        utils.String(accountName),
+		} 
+	} else {
+		return &streamingjobs.JobStorageAccount{
+			AuthenticationMode: utils.ToPtr(streamingjobs.AuthenticationMode(authenticationMode)),
+			AccountName:        utils.String(accountName),
+			AccountKey:         utils.String(accountKey),
+		}
 	}
 }
 

internal/services/streamanalytics/stream_analytics_job_resource_test.go

@@ -149,6 +149,27 @@ func TestAccStreamAnalyticsJob_jobStorageAccount(t *testing.T) {
 	})
 }
 
+func TestAccStreamAnalyticsJob_jobStorageAccount_Msi(t *testing.T) {
+	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_job", "test")
+	r := StreamAnalyticsJobResource{}
+
+	data.ResourceSequentialTest(t, r, []acceptance.TestStep{
+		{
+			Config: r.jobStorageAccount(data),
+			Check: acceptance.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		{
+			Config: r.jobStorageAccount_Msi(data),
+			Check: acceptance.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		data.ImportStep(),
+	})
+}
+
 func (r StreamAnalyticsJobResource) Exists(ctx context.Context, client *clients.Client, state *pluginsdk.InstanceState) (*bool, error) {
 	id, err := streamingjobs.ParseStreamingJobID(state.ID)
 	if err != nil {
@@ -422,3 +443,47 @@ QUERY
 }
 `, data.RandomInteger, data.RandomString, data.Locations.Primary, data.RandomInteger)
 }
+
+func (r StreamAnalyticsJobResource) jobStorageAccount_Msi(data acceptance.TestData) string {
+	return fmt.Sprintf(`
+provider "azurerm" {
+  features {}
+}
+
+resource "azurerm_resource_group" "test" {
+  name     = "acctestRG-%d"
+  location = "%[3]s"
+}
+
+resource "azurerm_storage_account" "test" {
+  name                     = "acctestacc%[2]s"
+  resource_group_name      = azurerm_resource_group.test.name
+  location                 = azurerm_resource_group.test.location
+  account_tier             = "Standard"
+  account_replication_type = "LRS"
+}
+
+resource "azurerm_stream_analytics_job" "test" {
+  name                   = "acctestjob-%[4]d"
+  resource_group_name    = azurerm_resource_group.test.name
+  location               = azurerm_resource_group.test.location
+  streaming_units        = 3
+  content_storage_policy = "JobStorageAccount"
+  job_storage_account {
+    authentication_mode = "Msi"
+    account_name        = azurerm_storage_account.test.name
+  }
+
+  tags = {
+    environment = "Test"
+  }
+
+  transformation_query = <<QUERY
+    SELECT *
+    INTO [YourOutputAlias]
+    FROM [YourInputAlias]
+QUERY
+
+}
+`, data.RandomInteger, data.RandomString, data.Locations.Primary, data.RandomInteger)
+}

website/docs/r/stream_analytics_job.html.markdown

@@ -87,11 +87,15 @@ The following arguments are supported:
 
 A `job_storage_account` block supports the following:
 
-* `authentication_mode` - (Optional) The authentication mode of the storage account. The only supported value is `ConnectionString`. Defaults to `ConnectionString`.
+* `authentication_mode` - (Optional) The authentication mode of the storage account. Possibles values are `ConnectionString`, `Msi`. Defaults to `ConnectionString`.
+
+-> **NOTE:** `authentication_mode` must be set to `ConnectionString` when creating a new job and can only be changed afterwards to `Msi`. This is because the job has to exists in order to authenticate against the storage account.
 
 * `account_name` - (Required) The name of the Azure storage account.
 
-* `account_key` - (Required) The account key for the Azure storage account.
+* `account_key` - (Optional) The account key for the Azure storage account.
+
+-> **NOTE:** `account_key` must be set when `authentication_mode` is `ConnectionString`.
 
 ---