New Resources: Stream Analytics Stream Inputs

azurerm/config.go

@@ -66,6 +66,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/services/servicebus/mgmt/2017-04-01/servicebus"
 	"github.com/Azure/azure-sdk-for-go/services/servicefabric/mgmt/2018-02-01/servicefabric"
 	"github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2018-02-01/storage"
+	"github.com/Azure/azure-sdk-for-go/services/streamanalytics/mgmt/2016-03-01/streamanalytics"
 	"github.com/Azure/azure-sdk-for-go/services/trafficmanager/mgmt/2018-04-01/trafficmanager"
 	"github.com/Azure/azure-sdk-for-go/services/web/mgmt/2018-02-01/web"
 
@@ -324,8 +325,7 @@ type ArmClient struct {
 
 	// Scheduler
 	schedulerJobCollectionsClient scheduler.JobCollectionsClient //nolint: megacheck
-
-	schedulerJobsClient scheduler.JobsClient //nolint: megacheck
+	schedulerJobsClient           scheduler.JobsClient           //nolint: megacheck
 
 	// Search
 	searchServicesClient  search.ServicesClient
@@ -353,6 +353,13 @@ type ArmClient struct {
 	storageServiceClient storage.AccountsClient
 	storageUsageClient   storage.UsageClient
 
+	// Stream Analytics
+	streamAnalyticsFunctionsClient       streamanalytics.FunctionsClient
+	streamAnalyticsJobsClient            streamanalytics.StreamingJobsClient
+	streamAnalyticsInputsClient          streamanalytics.InputsClient
+	streamAnalyticsOutputsClient         streamanalytics.OutputsClient
+	streamAnalyticsTransformationsClient streamanalytics.TransformationsClient
+
 	// Traffic Manager
 	trafficManagerGeographialHierarchiesClient trafficmanager.GeographicHierarchiesClient
 	trafficManagerProfilesClient               trafficmanager.ProfilesClient
@@ -514,6 +521,7 @@ func getArmClient(c *authentication.Config, skipProviderRegistration bool, partn
 	client.registerSchedulerClients(endpoint, c.SubscriptionID, auth)
 	client.registerSignalRClients(endpoint, c.SubscriptionID, auth)
 	client.registerStorageClients(endpoint, c.SubscriptionID, auth)
+	client.registerStreamAnalyticsClients(endpoint, c.SubscriptionID, auth)
 	client.registerTrafficManagerClients(endpoint, c.SubscriptionID, auth)
 	client.registerWebClients(endpoint, c.SubscriptionID, auth)
 
@@ -1318,6 +1326,28 @@ func (c *ArmClient) registerStorageClients(endpoint, subscriptionId string, auth
 	c.storageUsageClient = usageClient
 }
 
+func (c *ArmClient) registerStreamAnalyticsClients(endpoint, subscriptionId string, auth autorest.Authorizer) {
+	functionsClient := streamanalytics.NewFunctionsClientWithBaseURI(endpoint, subscriptionId)
+	c.configureClient(&functionsClient.Client, auth)
+	c.streamAnalyticsFunctionsClient = functionsClient
+
+	jobsClient := streamanalytics.NewStreamingJobsClientWithBaseURI(endpoint, subscriptionId)
+	c.configureClient(&jobsClient.Client, auth)
+	c.streamAnalyticsJobsClient = jobsClient
+
+	inputsClient := streamanalytics.NewInputsClientWithBaseURI(endpoint, subscriptionId)
+	c.configureClient(&inputsClient.Client, auth)
+	c.streamAnalyticsInputsClient = inputsClient
+
+	outputsClient := streamanalytics.NewOutputsClientWithBaseURI(endpoint, subscriptionId)
+	c.configureClient(&outputsClient.Client, auth)
+	c.streamAnalyticsOutputsClient = outputsClient
+
+	transformationsClient := streamanalytics.NewTransformationsClientWithBaseURI(endpoint, subscriptionId)
+	c.configureClient(&transformationsClient.Client, auth)
+	c.streamAnalyticsTransformationsClient = transformationsClient
+}
+
 func (c *ArmClient) registerTrafficManagerClients(endpoint, subscriptionId string, auth autorest.Authorizer) {
 	endpointsClient := trafficmanager.NewEndpointsClientWithBaseURI(endpoint, c.subscriptionId)
 	c.configureClient(&endpointsClient.Client, auth)

azurerm/data_source_stream_analytics_job.go

@@ -0,0 +1,130 @@
+package azurerm
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/terraform/helper/schema"
+	"github.com/terraform-providers/terraform-provider-azurerm/azurerm/utils"
+)
+
+func dataSourceArmStreamAnalyticsJob() *schema.Resource {
+	return &schema.Resource{
+		Read: dataSourceArmStreamAnalyticsJobRead,
+
+		Schema: map[string]*schema.Schema{
+			"name": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"resource_group_name": resourceGroupNameForDataSourceSchema(),
+
+			"location": locationForDataSourceSchema(),
+
+			"compatibility_level": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"data_locale": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"events_late_arrival_max_delay_in_seconds": {
+				Type:     schema.TypeInt,
+				Computed: true,
+			},
+
+			"events_out_of_order_max_delay_in_seconds": {
+				Type:     schema.TypeInt,
+				Computed: true,
+			},
+
+			"events_out_of_order_policy": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"job_id": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"output_error_policy": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"streaming_units": {
+				Type:     schema.TypeInt,
+				Computed: true,
+			},
+
+			"transformation_query": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+		},
+	}
+}
+
+func dataSourceArmStreamAnalyticsJobRead(d *schema.ResourceData, meta interface{}) error {
+	client := meta.(*ArmClient).streamAnalyticsJobsClient
+	transformationsClient := meta.(*ArmClient).streamAnalyticsTransformationsClient
+	ctx := meta.(*ArmClient).StopContext
+
+	name := d.Get("name").(string)
+	resourceGroup := d.Get("resource_group_name").(string)
+
+	resp, err := client.Get(ctx, resourceGroup, name, "")
+	if err != nil {
+		if utils.ResponseWasNotFound(resp.Response) {
+			return fmt.Errorf("Stream Analytics Job %q was not found in Resource Group %q!", name, resourceGroup)
+		}
+
+		return fmt.Errorf("Error retrieving Stream Analytics Job %q (Resource Group %q): %+v", name, resourceGroup, err)
+	}
+
+	transformation, err := transformationsClient.Get(ctx, resourceGroup, name, "Transformation")
+	if err != nil {
+		if utils.ResponseWasNotFound(resp.Response) {
+			return fmt.Errorf("Transformation for Stream Analytics Job %q was not found in Resource Group %q!", name, resourceGroup)
+		}
+
+		return fmt.Errorf("Error retrieving Transformation for Stream Analytics Job %q (Resource Group %q): %+v", name, resourceGroup, err)
+	}
+
+	d.SetId(*resp.ID)
+
+	d.Set("name", name)
+	d.Set("resource_group_name", resourceGroup)
+
+	if resp.Location != nil {
+		d.Set("location", azureRMNormalizeLocation(*resp.Location))
+	}
+
+	if props := resp.StreamingJobProperties; props != nil {
+		d.Set("compatibility_level", string(props.CompatibilityLevel))
+		d.Set("data_locale", props.DataLocale)
+		if props.EventsLateArrivalMaxDelayInSeconds != nil {
+			d.Set("events_late_arrival_max_delay_in_seconds", int(*props.EventsLateArrivalMaxDelayInSeconds))
+		}
+		if props.EventsOutOfOrderMaxDelayInSeconds != nil {
+			d.Set("events_out_of_order_max_delay_in_seconds", int(*props.EventsOutOfOrderMaxDelayInSeconds))
+		}
+		d.Set("events_out_of_order_policy", string(props.EventsOutOfOrderPolicy))
+		d.Set("job_id", props.JobID)
+		d.Set("output_error_policy", string(props.OutputErrorPolicy))
+	}
+
+	if props := transformation.TransformationProperties; props != nil {
+		if units := props.StreamingUnits; units != nil {
+			d.Set("streaming_units", int(*units))
+		}
+		d.Set("transformation_query", props.Query)
+	}
+
+	return nil
+}

azurerm/data_source_stream_analytics_job_test.go

@@ -0,0 +1,40 @@
+package azurerm
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/terraform/helper/resource"
+	"github.com/terraform-providers/terraform-provider-azurerm/azurerm/helpers/tf"
+)
+
+func TestAccDataSourceAzureRMStreamAnalyticsJob_basic(t *testing.T) {
+	dataSourceName := "data.azurerm_stream_analytics_job.test"
+	ri := tf.AccRandTimeInt()
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testCheckAzureRMStreamAnalyticsJobDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDataSourceAzureRMStreamAnalyticsJob_basic(ri, testLocation()),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttrSet(dataSourceName, "job_id"),
+				),
+			},
+		},
+	})
+}
+
+func testAccDataSourceAzureRMStreamAnalyticsJob_basic(rInt int, location string) string {
+	config := testAccAzureRMStreamAnalyticsJob_basic(rInt, location)
+	return fmt.Sprintf(`
+%s
+
+data "azurerm_stream_analytics_job" "test" {
+  name                = "${azurerm_stream_analytics_job.test.name}"
+  resource_group_name = "${azurerm_stream_analytics_job.test.resource_group_name}"
+}
+`, config)
+}