Merge pull request #13822 from hashicorp/f/stream-analytics-mssql-ref…

…erence-input

New resource `azurerm_stream_analytics_reference_input_mssql`

internal/services/streamanalytics/registration.go

@@ -36,6 +36,7 @@ func (r Registration) SupportedResources() map[string]*pluginsdk.Resource {
 		"azurerm_stream_analytics_output_servicebus_queue": resourceStreamAnalyticsOutputServiceBusQueue(),
 		"azurerm_stream_analytics_output_servicebus_topic": resourceStreamAnalyticsOutputServiceBusTopic(),
 		"azurerm_stream_analytics_reference_input_blob":    resourceStreamAnalyticsReferenceInputBlob(),
+		"azurerm_stream_analytics_reference_input_mssql":   resourceStreamAnalyticsReferenceMsSql(),
 		"azurerm_stream_analytics_stream_input_blob":       resourceStreamAnalyticsStreamInputBlob(),
 		"azurerm_stream_analytics_stream_input_eventhub":   resourceStreamAnalyticsStreamInputEventHub(),
 		"azurerm_stream_analytics_stream_input_iothub":     resourceStreamAnalyticsStreamInputIoTHub(),

internal/services/streamanalytics/stream_analytics_reference_input.go

@@ -0,0 +1,48 @@
+package streamanalytics
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Azure/azure-sdk-for-go/services/preview/streamanalytics/mgmt/2020-03-01-preview/streamanalytics"
+	"github.com/hashicorp/terraform-provider-azurerm/internal/clients"
+	"github.com/hashicorp/terraform-provider-azurerm/internal/services/streamanalytics/parse"
+	"github.com/hashicorp/terraform-provider-azurerm/internal/tf/pluginsdk"
+)
+
+func importStreamAnalyticsReferenceInput(expectType streamanalytics.TypeBasicReferenceInputDataSource) pluginsdk.ImporterFunc {
+	return func(ctx context.Context, d *pluginsdk.ResourceData, meta interface{}) (data []*pluginsdk.ResourceData, err error) {
+		id, err := parse.StreamInputID(d.Id())
+		if err != nil {
+			return nil, err
+		}
+
+		client := meta.(*clients.Client).StreamAnalytics.InputsClient
+		resp, err := client.Get(ctx, id.ResourceGroup, id.StreamingjobName, id.InputName)
+		if err != nil {
+			return nil, fmt.Errorf("retrieving %s: %+v", *id, err)
+		}
+
+		if props := resp.Properties; props != nil {
+			v, ok := props.AsReferenceInputProperties()
+			if !ok {
+				return nil, fmt.Errorf("converting properties to a Reference Input: %+v", err)
+			}
+
+			var actualType streamanalytics.TypeBasicReferenceInputDataSource
+
+			if inputMsSql, ok := v.Datasource.AsAzureSQLReferenceInputDataSource(); ok {
+				actualType = inputMsSql.Type
+			} else if inputBlob, ok := v.Datasource.AsBlobReferenceInputDataSource(); ok {
+				actualType = inputBlob.Type
+			} else {
+				return nil, fmt.Errorf("unable to convert input data source: %+v", v)
+			}
+
+			if actualType != expectType {
+				return nil, fmt.Errorf("stream analytics reference input has mismatched type, expected: %q, got %q", expectType, actualType)
+			}
+		}
+		return []*pluginsdk.ResourceData{d}, nil
+	}
+}

internal/services/streamanalytics/stream_analytics_reference_input_blob_resource.go

@@ -23,10 +23,11 @@ func resourceStreamAnalyticsReferenceInputBlob() *pluginsdk.Resource {
 		Read:   resourceStreamAnalyticsReferenceInputBlobRead,
 		Update: resourceStreamAnalyticsReferenceInputBlobUpdate,
 		Delete: resourceStreamAnalyticsReferenceInputBlobDelete,
-		Importer: pluginsdk.ImporterValidatingResourceId(func(id string) error {
+
+		Importer: pluginsdk.ImporterValidatingResourceIdThen(func(id string) error {
 			_, err := parse.StreamInputID(id)
 			return err
-		}),
+		}, importStreamAnalyticsReferenceInput(streamanalytics.TypeBasicReferenceInputDataSourceTypeMicrosoftStorageBlob)),
 
 		Timeouts: &pluginsdk.ResourceTimeout{
 			Create: pluginsdk.DefaultTimeout(30 * time.Minute),
@@ -101,17 +102,17 @@ func resourceStreamAnalyticsReferenceInputBlobCreate(d *pluginsdk.ResourceData,
 	defer cancel()
 
 	log.Printf("[INFO] preparing arguments for Azure Stream Analytics Reference Input Blob creation.")
-	resourceId := parse.NewStreamInputID(subscriptionId, d.Get("resource_group_name").(string), d.Get("stream_analytics_job_name").(string), d.Get("name").(string))
+	id := parse.NewStreamInputID(subscriptionId, d.Get("resource_group_name").(string), d.Get("stream_analytics_job_name").(string), d.Get("name").(string))
 	if d.IsNewResource() {
-		existing, err := client.Get(ctx, resourceId.ResourceGroup, resourceId.StreamingjobName, resourceId.InputName)
+		existing, err := client.Get(ctx, id.ResourceGroup, id.StreamingjobName, id.InputName)
 		if err != nil {
 			if !utils.ResponseWasNotFound(existing.Response) {
-				return fmt.Errorf("checking for presence of existing %s: %+v", resourceId, err)
+				return fmt.Errorf("checking for presence of existing %s: %+v", id, err)
 			}
 		}
 
 		if existing.ID != nil && *existing.ID != "" {
-			return tf.ImportAsExistsError("azurerm_stream_analytics_reference_input_blob", resourceId.ID())
+			return tf.ImportAsExistsError("azurerm_stream_analytics_reference_input_blob", id.ID())
 		}
 	}
 
@@ -120,11 +121,11 @@ func resourceStreamAnalyticsReferenceInputBlobCreate(d *pluginsdk.ResourceData,
 		return fmt.Errorf("creating the input props for resource creation: %v", err)
 	}
 
-	if _, err := client.CreateOrReplace(ctx, props, resourceId.ResourceGroup, resourceId.StreamingjobName, resourceId.InputName, "", ""); err != nil {
-		return fmt.Errorf("creating %s: %+v", resourceId, err)
+	if _, err := client.CreateOrReplace(ctx, props, id.ResourceGroup, id.StreamingjobName, id.InputName, "", ""); err != nil {
+		return fmt.Errorf("creating %s: %+v", id, err)
 	}
 
-	d.SetId(resourceId.ID())
+	d.SetId(id.ID())
 	return resourceStreamAnalyticsReferenceInputBlobRead(d, meta)
 }
 
@@ -145,7 +146,7 @@ func resourceStreamAnalyticsReferenceInputBlobUpdate(d *pluginsdk.ResourceData,
 	}
 
 	if _, err := client.Update(ctx, props, id.ResourceGroup, id.StreamingjobName, id.InputName, ""); err != nil {
-		return fmt.Errorf("updating %s: %+v", id, err)
+		return fmt.Errorf("updating %s: %+v", *id, err)
 	}
 
 	return resourceStreamAnalyticsReferenceInputBlobRead(d, meta)
@@ -164,7 +165,7 @@ func resourceStreamAnalyticsReferenceInputBlobRead(d *pluginsdk.ResourceData, me
 	resp, err := client.Get(ctx, id.ResourceGroup, id.StreamingjobName, id.InputName)
 	if err != nil {
 		if utils.ResponseWasNotFound(resp.Response) {
-			log.Printf("[DEBUG] %s was not found - removing from state!", id)
+			log.Printf("[DEBUG] %s was not found - removing from state!", *id)
 			d.SetId("")
 			return nil
 		}
@@ -217,7 +218,7 @@ func resourceStreamAnalyticsReferenceInputBlobDelete(d *pluginsdk.ResourceData,
 
 	if resp, err := client.Delete(ctx, id.ResourceGroup, id.StreamingjobName, id.InputName); err != nil {
 		if !response.WasNotFound(resp.Response) {
-			return fmt.Errorf("deleting %s: %+v", id, err)
+			return fmt.Errorf("deleting %s: %+v", *id, err)
 		}
 	}
 

internal/services/streamanalytics/stream_analytics_reference_input_blob_resource_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/hashicorp/terraform-provider-azurerm/internal/acceptance"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/acceptance/check"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/clients"
+	"github.com/hashicorp/terraform-provider-azurerm/internal/services/streamanalytics/parse"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/tf/pluginsdk"
 	"github.com/hashicorp/terraform-provider-azurerm/utils"
 )
@@ -96,16 +97,17 @@ func TestAccStreamAnalyticsReferenceInputBlob_requiresImport(t *testing.T) {
 }
 
 func (r StreamAnalyticsReferenceInputBlobResource) Exists(ctx context.Context, client *clients.Client, state *pluginsdk.InstanceState) (*bool, error) {
-	name := state.Attributes["name"]
-	jobName := state.Attributes["stream_analytics_job_name"]
-	resourceGroup := state.Attributes["resource_group_name"]
+	id, err := parse.StreamInputID(state.ID)
+	if err != nil {
+		return nil, err
+	}
 
-	resp, err := client.StreamAnalytics.InputsClient.Get(ctx, resourceGroup, jobName, name)
+	resp, err := client.StreamAnalytics.InputsClient.Get(ctx, id.ResourceGroup, id.StreamingjobName, id.InputName)
 	if err != nil {
 		if utils.ResponseWasNotFound(resp.Response) {
 			return utils.Bool(false), nil
 		}
-		return nil, fmt.Errorf("retrieving Stream Output %q (Stream Analytics Job %q / Resource Group %q): %+v", name, jobName, resourceGroup, err)
+		return nil, fmt.Errorf("retrieving (%s): %+v", *id, err)
 	}
 	return utils.Bool(true), nil
 }