CyberArk(client): add CyberArk snapshot conversion

- Introduced `convertDataReadings` to process `DataReading` objects into snapshots.
- Added support for extracting Kubernetes server version and dynamic resources.
- Updated `CyberArkClient` to use the new data conversion logic.
- Refactored `DiscoveryData` and `DynamicData` structures for better type safety.
- Replaced `unstructured.Unstructured` with `runtime.Object` in `Snapshot` fields.
- Enhanced `DataGathererDiscovery` and `DataGathererDynamic` to return strongly typed data.
- Added unit tests for new data extraction and conversion functions.

Signed-off-by: Richard Wall <[email protected]>

Author: wallrj-cyberark

api/datareading.go

@@ -3,6 +3,8 @@ package api
 import (
 	"encoding/json"
 	"time"
+
+	"k8s.io/apimachinery/pkg/version"
 )
 
 // DataReadingsPost is the payload in the upload request.
@@ -48,3 +50,18 @@ func (v GatheredResource) MarshalJSON() ([]byte, error) {
 
 	return json.Marshal(data)
 }
+
+// DynamicData is the DataReading.Data returned by the k8s.DataGathererDynamic
+// gatherer
+type DynamicData struct {
+	// Items is a list of GatheredResource
+	Items []*GatheredResource `json:"items"`
+}
+
+// DiscoveryData is the DataReading.Data returned by the k8s.ConfigDiscovery
+// gatherer
+type DiscoveryData struct {
+	// ServerVersion is the version information of the k8s apiserver
+	// See https://godoc.org/k8s.io/apimachinery/pkg/version#Info
+	ServerVersion *version.Info `json:"server_version"`
+}

pkg/client/client_cyberark.go

@@ -5,10 +5,12 @@ import (
 	"fmt"
 	"net/http"
 
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+
 	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/internal/cyberark"
 	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
-	"github.com/jetstack/preflight/pkg/version"
 )
 
 // CyberArkClient is a client for publishing data readings to CyberArk's discoverycontext API.
@@ -48,15 +50,144 @@ func (o *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, readin
 	if err != nil {
 		return fmt.Errorf("while initializing data upload client: %s", err)
 	}
+	var snapshot dataupload.Snapshot
+	if err := convertDataReadings(defaultExtractorFunctions, readings, &snapshot); err != nil {
+		return fmt.Errorf("while converting data readings: %s", err)
+	}
+	// Temporary hard coded cluster ID.
+	// TODO(wallrj): The clusterID will eventually be extracted from the supplied readings.
+	snapshot.ClusterID = "success-cluster-id"
 
-	err = datauploadClient.PutSnapshot(ctx, dataupload.Snapshot{
-		// Temporary hard coded cluster ID.
-		// TODO(wallrj): The clusterID will eventually be extracted from the supplied readings.
-		ClusterID:    "success-cluster-id",
-		AgentVersion: version.PreflightVersion,
-	})
+	err = datauploadClient.PutSnapshot(ctx, snapshot)
 	if err != nil {
 		return fmt.Errorf("while uploading snapshot: %s", err)
 	}
 	return nil
 }
+
+// extractServerVersionFromReading converts the opaque data from a DiscoveryData
+// data reading to allow access to the Kubernetes version fields within.
+func extractServerVersionFromReading(reading *api.DataReading, target *string) error {
+	if reading == nil {
+		return fmt.Errorf("programmer mistake: the DataReading must not be nil")
+	}
+	data, ok := reading.Data.(*api.DiscoveryData)
+	if !ok {
+		return fmt.Errorf(
+			"programmer mistake: the DataReading must have data type *api.DiscoveryData. "+
+				"This DataReading (%s) has data type %T", reading.DataGatherer, reading.Data)
+	}
+	if data.ServerVersion == nil {
+		return nil
+	}
+	*target = data.ServerVersion.GitVersion
+	return nil
+}
+
+// extractResourceListFromReading converts the opaque data from a DynamicData
+// data reading to runtime.Object resources, to allow access to the metadata and
+// other kubernetes API fields.
+func extractResourceListFromReading(reading *api.DataReading, target *[]runtime.Object) error {
+	if reading == nil {
+		return fmt.Errorf("programmer mistake: the DataReading must not be nil")
+	}
+	data, ok := reading.Data.(*api.DynamicData)
+	if !ok {
+		return fmt.Errorf(
+			"programmer mistake: the DataReading must have data type *api.DynamicData. "+
+				"This DataReading (%s) has data type %T", reading.DataGatherer, reading.Data)
+	}
+	resources := make([]runtime.Object, len(data.Items))
+	for i, item := range data.Items {
+		if resource, ok := item.Resource.(runtime.Object); ok {
+			resources[i] = resource
+		} else {
+			return fmt.Errorf(
+				"programmer mistake: the DynamicData items must have Resource type runtime.Object. "+
+					"This item (%d) has Resource type %T", i, item.Resource)
+		}
+	}
+	*target = resources
+	return nil
+}
+
+var defaultExtractorFunctions = map[string]func(*api.DataReading, *dataupload.Snapshot) error{
+	"ark/discovery": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractServerVersionFromReading(r, &s.K8SVersion)
+	},
+	"ark/secrets": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Secrets)
+	},
+	"ark/serviceaccounts": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.ServiceAccounts)
+	},
+	"ark/roles": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Roles)
+	},
+	"ark/clusterroles": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.ClusterRoles)
+	},
+	"ark/rolebindings": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.RoleBindings)
+	},
+	"ark/clusterrolebindings": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.ClusterRoleBindings)
+	},
+	"ark/jobs": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Jobs)
+	},
+	"ark/cronjobs": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.CronJobs)
+	},
+	"ark/deployments": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Deployments)
+	},
+	"ark/statefulsets": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Statefulsets)
+	},
+	"ark/daemonsets": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Daemonsets)
+	},
+	"ark/pods": func(r *api.DataReading, s *dataupload.Snapshot) error {
+		return extractResourceListFromReading(r, &s.Pods)
+	},
+}
+
+// convertDataReadings processes a list of DataReadings using the provided
+// extractor functions to populate the fields of the target snapshot.
+// It ensures that all expected data gatherers are handled and that there are
+// no unhandled data gatherers. If any discrepancies are found, or if any
+// extractor function returns an error, it returns an error.
+// The extractorFunctions map should contain functions for each expected
+// DataGatherer name, which will be called with the corresponding DataReading
+// and the target snapshot to populate the relevant fields.
+func convertDataReadings[T any](
+	extractorFunctions map[string]func(*api.DataReading, *T) error,
+	readings []*api.DataReading,
+	target *T,
+) error {
+	expectedDataGatherers := sets.StringKeySet(extractorFunctions)
+	unhandledDataGatherers := sets.New[string]()
+	missingDataGatherers := sets.New[string](expectedDataGatherers.UnsortedList()...)
+	for _, reading := range readings {
+		dataGathererName := reading.DataGatherer
+		extractFunc, found := extractorFunctions[dataGathererName]
+		if !found {
+			unhandledDataGatherers.Insert(dataGathererName)
+			continue
+		}
+		missingDataGatherers.Delete(dataGathererName)
+		// Call the extractor function to populate the relevant field in the target snapshot.
+		if err := extractFunc(reading, target); err != nil {
+			return fmt.Errorf("while extracting data reading %s: %s", dataGathererName, err)
+		}
+	}
+	if missingDataGatherers.Len() > 0 || unhandledDataGatherers.Len() > 0 {
+		return fmt.Errorf(
+			"unexpected data gatherers, missing: %v, unhandled: %v",
+			sets.List(missingDataGatherers),
+			sets.List(unhandledDataGatherers),
+		)
+	}
+	return nil
+}