feat: azure resources id management improvements

docs/how-to/030_azure-source.md

@@ -55,15 +55,44 @@ Using the `*_CONNECTION_STRING` variables is the preferred methods that will als
 setup the least privileges to the service account responsible to retrieve information from the
 REST APIs.
 
+## Resource identifiers
+
+Microsoft Azure does not guarantee the letter case of the resource IDs it returns: the same
+resource can arrive with `resourceGroups` from the resource graph APIs and with `resourcegroups`
+from the resource provider that answers the EventHub driven read, and the provider and type
+segments vary in the same way. Because the mappings hash the ID to build the Catalog identifier,
+every casing difference would create a duplicate item instead of updating the existing one.
+
+To prevent that the source normalises every resource before handing it to the mapper:
+
+- `id` is lowercased in full
+- `type` is set to the resource type exactly as the mapping file declares it
+
+The two values are therefore identical for `ibdm sync azure` and `ibdm run azure`, which makes
+`{{ .id | sha256sum }}` a stable identifier and lets a delete event target the item a previous
+import created.
+
+This is also in part suggested by Azure, since it is stated that various APIs can return names with different casing,
+therefore in order to perform meaningful matches a case-insensitive comparison is recommended.
+For a more in-depth explanation refer to [Naming rules and restrictions for Azure resources].
+
+### Consequences for the mappings and items
+
+`id` is lowercase and could no longer match the casing shown in the Azure portal.
+Use `.name` wherever available to display casing matters, its availability is dependant on the specific resource APIs.
+The spelling Azure reported, if needed, is written to the source logs at the `Debug` level whenever it differs from the normalised value.
+
 ## Authentication
 
 The source is using the [`DefaultAzureCredential` chain of authentication] so you can setup
 your preferred method of login.  
 This authentication will be used for reading data from the REST APIs so it will need the read
 permissions on the resources you want to import.
+Both `sync` and `run` modes use APIs to fetch the full resource, for this reason an authentication method of choice is always needed.
 
 If you choose to don’t use the `*_CONNECTION_STING` variables the same authentication will be used
 to receive data from the configured EventHub and to manage object inside the StorageAccount blob
 storage.
 
 [`DefaultAzureCredential` chain of authentication]: https://learn.microsoft.com/en-gb/azure/developer/go/sdk/authentication/credential-chains#defaultazurecredential-overview
+[Naming rules and restrictions for Azure resources]: https://learn.microsoft.com/en-us/azure/azure-resource-manager/management/resource-name-rules

docs/mappings/azure/cognitiveaccounts.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     kind: "{{ .kind }}"

docs/mappings/azure/containerapps.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     runningStatus: "{{ .properties.runningStatus | lower }}"

docs/mappings/azure/managedclusters.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     currentKubernetesVersion: "{{ .properties.currentKubernetesVersion | quote }}"
@@ -34,7 +35,7 @@ mappings:
   extra:
     - apiVersion: mia-platform.eu/v1
       itemFamily: relationships
-      deletePolicy: "cascade"
+      deletePolicy: "none"
       identifier: |-
         {{ $src := printf "urn:mia-platform-catalog:console.mia-platform.eu:v1:Cluster:%s" (printf "%s" .properties.fqdn | sha256sum) -}}
         {{- $type := "urn:mia-platform-catalog:mia-platform.eu:v1:RelationshipType:dependency.mia-platform.eu" -}}

docs/mappings/azure/postgresqldbs.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     state: "{{ .properties.state | lower }}"
     version: "{{ .properties.version }}"

docs/mappings/azure/resourcegroups.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     tags: |-

docs/mappings/azure/storageaccounts.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     allowBlobPublicAccess: "{{ get \"allowBlobPublicAccess\" .properties false }}"
     publicNetworkAccess: "{{ get \"publicNetworkAccess\" .properties \"Disabled\" | lower }}"

docs/mappings/azure/subscriptions.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     displayName: "{{ get \"displayName\" . \"\" | quote }}"
     state: "{{ .properties.state | lower }}"
     tags: |-

docs/mappings/azure/virtualmachines.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     vmSize: "{{ .properties.hardwareProfile.vmSize }}"

docs/mappings/azure/virtualnetworks.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     provisioningState: "{{ .properties.provisioningState | lower }}"
     addressPrefixes: "{{ .properties.addressSpace.addressPrefixes | toJSON }}"

docs/mappings/azure/websites.yaml

@@ -10,6 +10,7 @@ mappings:
     title: "{{ .name }}"
   spec:
     name: "{{ .name }}"
+    id: "{{ .id }}"
     location: "{{ .location | lower }}"
     state: "{{ .properties.state | lower }}"
     kind: "{{ .kind }}"

internal/source/azure/azure.go

@@ -10,7 +10,6 @@ import (
 	"fmt"
 	"maps"
 	"slices"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -143,7 +142,7 @@ func (s *Source) StartSyncProcess(ctx context.Context, typesToFilter map[string]
 			if data, ok := response.Data.([]any); ok {
 				for _, item := range data {
 					if values, ok := item.(map[string]any); ok {
-						values["type"] = resType // ensure type is case-normalized, and resourceGroup is normalized too
+						normalizeResourceValues(logger, values, resType)
 						dataChannel <- source.Data{
 							Type:      resType,
 							Operation: source.DataOperationUpsert,
@@ -224,18 +223,21 @@ func partitionEventHandler(client *armresources.Client, typesToFilter map[string
 				continue
 			}
 
-			if filterBasedOnResourceID(resID, typesSlice) {
-				logger.Debug("skipping event based on type", "resourceID", resID.ResourceType.String())
+			// the subject can spell the resource type with any casing, so resolve the configured
+			// key once and use it for the apiVersion lookup and for every emitted value.
+			resourceType, ok := configuredResourceType(typesSlice, resID.ResourceType.String())
+			if !ok {
+				logger.Debug("skipping event based on type", "resourceType", resID.ResourceType.String())
 				continue
 			}
 
-			apiVersion, ok := typesToFilter[resID.ResourceType.String()]["apiVersion"].(string)
+			apiVersion, ok := typesToFilter[resourceType][apiVersionKey].(string)
 			if !ok {
-				logger.Debug("skipping event with missing apiVersion", "resourceID", resID.ResourceType.String())
+				logger.Debug("skipping event with missing apiVersion", "resourceType", resourceType)
 				continue
 			}
 
-			logger.Trace("handling resource", "resourceID", resID.ResourceType.String(), "eventType", envelope.Type, "apiVersion", apiVersion)
+			logger.Trace("handling resource", "resourceType", resourceType, "eventType", envelope.Type, "apiVersion", apiVersion)
 			switch envelope.Type {
 			case azsystemevents.TypeResourceWriteSuccess:
 				logger.Trace("request resource data from azure", "resourceID", *envelope.Subject)
@@ -255,25 +257,25 @@ func partitionEventHandler(client *armresources.Client, typesToFilter map[string
 					continue
 				}
 
+				normalizeResourceValues(logger, values, resourceType)
 				dataChannel <- source.Data{
-					Type:      resID.ResourceType.String(),
+					Type:      resourceType,
 					Operation: source.DataOperationUpsert,
 					Time:      *envelope.Time,
 					Values:    values,
 				}
 			case azsystemevents.TypeResourceDeleteSuccess:
-				logger.Trace("we have to delete something", "resourceID", resID.ResourceType.String())
+				logger.Trace("deleting resource", "resourceType", resourceType)
+				values := map[string]any{idKey: resID.String()}
+				normalizeResourceValues(logger, values, resourceType)
 				dataChannel <- source.Data{
-					Type:      resID.ResourceType.String(),
+					Type:      resourceType,
 					Operation: source.DataOperationDelete,
 					Time:      *envelope.Time,
-					Values: map[string]any{
-						"id":   resID.String(),
-						"type": resID.ResourceType.String(),
-					},
+					Values:    values,
 				}
 			default:
-				logger.Trace("skipping resource", "resourceID", resID.ResourceType.String(), "eventType", envelope.Type, "apiVersion", apiVersion)
+				logger.Trace("skipping resource", "resourceType", resourceType, "eventType", envelope.Type, "apiVersion", apiVersion)
 			}
 		}
 	}
@@ -303,14 +305,6 @@ func resourceIDFromSubject(subject *string) (*arm.ResourceID, error) {
 	return arm.ParseResourceID(*subject)
 }
 
-// filterBasedOnResourceID checks if the resource type is in the typesToFilter slice.
-func filterBasedOnResourceID(resID *arm.ResourceID, typesToFilter []string) bool {
-	resourceType := resID.ResourceType.String()
-	return !slices.ContainsFunc(typesToFilter, func(s string) bool {
-		return strings.EqualFold(s, resourceType)
-	})
-}
-
 // Close implement source.ClosableSource.
 func (s *Source) Close(ctx context.Context, _ time.Duration) error {
 	log := logger.FromContext(ctx).WithName(logName)

internal/source/azure/azure_identifier_test.go

@@ -0,0 +1,129 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package azure
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
+	fakeazcore "github.com/Azure/azure-sdk-for-go/sdk/azcore/fake"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/resources/armresources/v3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mia-platform/ibdm/internal/source"
+)
+
+const testSubscriptionID = "00000000-0000-0000-0000-000000000000"
+
+// eventDataManagedClusterCanonicalDeleteBody deletes the same managed cluster the sync and the
+// stream upsert paths import. Its subject uses a lowercase resourcegroups literal that
+// arm.ParseResourceID rewrites to camelCase, so the delete path builds subjectManagedClusterID.
+var eventDataManagedClusterCanonicalDeleteBody = json.RawMessage(`[
+{
+	"id": "00000000-0000-0000-0000-000000000000",
+	"source": "/subscriptions/00000000-0000-0000-0000-000000000000",
+	"specversion": "1.0",
+	"type": "Microsoft.Resources.ResourceDeleteSuccess",
+	"subject": "/subscriptions/00000000-0000-0000-0000-000000000000/resourcegroups/my-rg/providers/Microsoft.ContainerService/managedClusters/my-cluster",
+	"time": "2020-01-01T00:00:00.0000000Z"
+}]`)
+
+// TestAllPathsEmitTheSameIdentifier drives the three ingestion paths for one managed cluster and
+// checks that they agree on the values the mappings hash into the Catalog identifier. Azure feeds
+// each path a different spelling of the ID, so before the normalization the sync upsert and the
+// stream upsert produced two Catalog items and the stream delete targeted neither of them.
+func TestAllPathsEmitTheSameIdentifier(t *testing.T) {
+	t.Parallel()
+
+	// the three spellings the fakes reproduce: Resource Graph answers with a camelCase
+	// resourceGroups literal, the resource provider body lowercases it, and the delete path rebuilds
+	// the ID from the event subject. Without this divergence the test would be vacuous.
+	require.NotEqual(t, graphManagedClusterID, bodyManagedClusterID,
+		"the sync and stream fixtures must disagree on casing")
+	require.Equal(t, graphManagedClusterID, subjectManagedClusterID,
+		"the delete path is expected to rebuild the Resource Graph spelling")
+
+	paths := map[string]source.Data{
+		"sync upsert":   syncedManagedCluster(t),
+		"stream upsert": streamedManagedCluster(t, eventDataManagedClusterWriteBody),
+		"stream delete": streamedManagedCluster(t, eventDataManagedClusterCanonicalDeleteBody),
+	}
+
+	assert.Equal(t, source.DataOperationUpsert, paths["sync upsert"].Operation)
+	assert.Equal(t, source.DataOperationUpsert, paths["stream upsert"].Operation)
+	assert.Equal(t, source.DataOperationDelete, paths["stream delete"].Operation)
+
+	for pathName, data := range paths {
+		assert.Equal(t, normalizedManagedClusterID, data.Values[idKey], pathName)
+		assert.Equal(t, managedClustersType, data.Values[typeKey], pathName)
+		assert.Equal(t, managedClustersType, data.Type, pathName)
+	}
+}
+
+// syncedManagedCluster runs a sync process against the Resource Graph fake and returns the data
+// emitted for the my-cluster managed cluster.
+func syncedManagedCluster(t *testing.T) source.Data {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 1*time.Second)
+	t.Cleanup(cancel)
+
+	azureSource := &Source{
+		config: config{
+			SubscriptionID: testSubscriptionID,
+			clientOptions: &arm.ClientOptions{
+				ClientOptions: policy.ClientOptions{
+					Transport: fakeResourceGraphTransport(t),
+				},
+			},
+			azureCredentials: &fakeazcore.TokenCredential{},
+		},
+	}
+
+	dataChannel := make(chan source.Data, 10)
+	require.NoError(t, azureSource.StartSyncProcess(ctx, map[string]source.Extra{managedClustersType: nil}, dataChannel))
+	close(dataChannel)
+
+	for data := range dataChannel {
+		if data.Values["name"] == "my-cluster" {
+			return data
+		}
+	}
+
+	require.FailNow(t, "the sync process did not emit the managed cluster")
+	return source.Data{}
+}
+
+// streamedManagedCluster feeds body to the event handler and returns the single emitted data.
+func streamedManagedCluster(t *testing.T, body json.RawMessage) source.Data {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 1*time.Second)
+	t.Cleanup(cancel)
+
+	client, err := armresources.NewClient(testSubscriptionID, &fakeazcore.TokenCredential{}, &arm.ClientOptions{
+		ClientOptions: policy.ClientOptions{
+			Transport: fakeClientTransport(t),
+		},
+	})
+	require.NoError(t, err)
+
+	dataChannel := make(chan source.Data, 10)
+	handler := partitionEventHandler(client, map[string]source.Extra{
+		managedClustersType: {apiVersionKey: managedClustersAPIVersion},
+	}, dataChannel)
+
+	handler(ctx, &azeventhubs.ReceivedEventData{EventData: azeventhubs.EventData{Body: body}})
+	close(dataChannel)
+
+	data, ok := <-dataChannel
+	require.True(t, ok, "the event handler did not emit any data")
+	return data
+}