diff --git a/api/datareading.go b/api/datareading.go
index 54637f3c..1c08701a 100644
--- a/api/datareading.go
+++ b/api/datareading.go
@@ -1,8 +1,11 @@
 package api
 
 import (
+	"bytes"
 	"encoding/json"
 	"time"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
 // DataReadingsPost is the payload in the upload request.
@@ -28,8 +31,8 @@ type DataReading struct {
 type GatheredResource struct {
 	// Resource is a reference to a k8s object that was found by the informer
 	// should be of type unstructured.Unstructured, raw Object
-	Resource  interface{}
-	DeletedAt Time
+	Resource  interface{} `json:"resource"`
+	DeletedAt Time        `json:"deleted_at,omitempty"`
 }
 
 func (v GatheredResource) MarshalJSON() ([]byte, error) {
@@ -48,3 +51,20 @@ func (v GatheredResource) MarshalJSON() ([]byte, error) {
 
 	return json.Marshal(data)
 }
+
+func (v *GatheredResource) UnmarshalJSON(data []byte) error {
+	var tmpResource struct {
+		Resource  *unstructured.Unstructured `json:"resource"`
+		DeletedAt Time        `json:"deleted_at,omitempty"`
+	}
+
+	d := json.NewDecoder(bytes.NewReader(data))
+	d.DisallowUnknownFields()
+
+	if err := d.Decode(&tmpResource); err != nil {
+		return err
+	}
+	v.Resource = tmpResource.Resource
+	v.DeletedAt = tmpResource.DeletedAt
+	return nil
+}
diff --git a/deploy/charts/venafi-kubernetes-agent/templates/deployment.yaml b/deploy/charts/venafi-kubernetes-agent/templates/deployment.yaml
index f37e8a16..270df61a 100644
--- a/deploy/charts/venafi-kubernetes-agent/templates/deployment.yaml
+++ b/deploy/charts/venafi-kubernetes-agent/templates/deployment.yaml
@@ -49,6 +49,26 @@ spec:
             valueFrom:
               fieldRef:
                 fieldPath: spec.nodeName
+          - name: ARK_USERNAME
+            valueFrom:
+              secretKeyRef:
+                name: {{ .Values.authentication.secretName }}
+                key: ARK_USERNAME
+          - name: ARK_SECRET
+            valueFrom:
+              secretKeyRef:
+                name: {{ .Values.authentication.secretName }}
+                key: ARK_SECRET
+          - name: ARK_PLATFORM_DOMAIN
+            valueFrom:
+              secretKeyRef:
+                name: {{ .Values.authentication.secretName }}
+                key: ARK_PLATFORM_DOMAIN
+          - name: ARK_SUBDOMAIN
+            valueFrom:
+              secretKeyRef:
+                name: {{ .Values.authentication.secretName }}
+                key: ARK_SUBDOMAIN
           {{- with .Values.http_proxy }}
           - name: HTTP_PROXY
             value: {{ . }}
@@ -71,18 +91,8 @@ spec:
             - "agent"
             - "-c"
             - "/etc/venafi/agent/config/{{ default "config.yaml" .Values.config.configmap.key }}"
-            {{- if .Values.authentication.venafiConnection.enabled }}
-            - --venafi-connection
-            - {{ .Values.authentication.venafiConnection.name | quote }}
-            - --venafi-connection-namespace
-            - {{ .Values.authentication.venafiConnection.namespace | quote }}
-            {{- else }}
-            - "--client-id"
-            - {{ .Values.config.clientId | quote }}
-            - "--private-key-path"
-            - "/etc/venafi/agent/key/{{ .Values.authentication.secretKey }}"
-            {{- end }}
-            - --venafi-cloud
+            - --log-level=6
+            - --machine-hub
             {{- if .Values.metrics.enabled }}
             - --enable-metrics
             {{- end }}
@@ -95,11 +105,6 @@ spec:
             - name: config
               mountPath: "/etc/venafi/agent/config"
               readOnly: true
-            {{- if not .Values.authentication.venafiConnection.enabled }}
-            - name: credentials
-              mountPath: "/etc/venafi/agent/key"
-              readOnly: true
-            {{- end }}
             {{- with .Values.volumeMounts }}
             {{- toYaml . | nindent 12 }}
             {{- end }}
@@ -137,12 +142,6 @@ spec:
           configMap:
             name: {{ default "agent-config" .Values.config.configmap.name }}
             optional: false
-        {{- if not .Values.authentication.venafiConnection.enabled }}
-        - name: credentials
-          secret:
-            secretName: {{ .Values.authentication.secretName }}
-            optional: false
-        {{- end }}
         {{- with .Values.volumes }}
         {{- toYaml . | nindent 8 }}
         {{- end }}
diff --git a/hack/e2e/ca/config.yaml b/hack/e2e/ca/config.yaml
new file mode 100644
index 00000000..9644d534
--- /dev/null
+++ b/hack/e2e/ca/config.yaml
@@ -0,0 +1,3 @@
+machineHub:
+  subdomain: tlskp-test
+  credentialsSecretName: todo-unused
diff --git a/hack/e2e/ca/test.sh b/hack/e2e/ca/test.sh
new file mode 100755
index 00000000..579b513f
--- /dev/null
+++ b/hack/e2e/ca/test.sh
@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+#
+set -o nounset
+set -o errexit
+set -o pipefail
+
+# CyberArk API configuration
+: ${ARK_USERNAME?}
+: ${ARK_SECRET?}
+: ${ARK_PLATFORM_DOMAIN?}
+: ${ARK_SUBDOMAIN?}
+
+# The base URL of the OCI registry used for Docker images and Helm charts
+# E.g. ttl.sh/6ee49a01-c8ba-493e-bae9-4d8567574b56
+: ${OCI_BASE?}
+
+k8s_namespace=cyberark
+
+script_dir=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+root_dir=$(cd "${script_dir}/../../.." && pwd)
+export TERM=dumb
+
+tmp_dir="$(mktemp -d /tmp/jetstack-secure.XXXXX)"
+
+pushd "${tmp_dir}"
+> release.env
+make -C "$root_dir" release \
+     OCI_SIGN_ON_PUSH=false \
+     oci_platforms=linux/amd64 \
+     oci_preflight_image_name=$OCI_BASE/images/venafi-agent \
+     helm_chart_image_name=$OCI_BASE/charts/venafi-kubernetes-agent \
+     GITHUB_OUTPUT="${tmp_dir}/release.env"
+source release.env
+
+kind create cluster || true
+kubectl create ns "$k8s_namespace" || true
+
+kubectl create secret generic agent-credentials \
+        --namespace "$k8s_namespace" \
+        --from-literal=ARK_USERNAME=$ARK_USERNAME \
+        --from-literal=ARK_SECRET=$ARK_SECRET \
+        --from-literal=ARK_PLATFORM_DOMAIN=$ARK_PLATFORM_DOMAIN \
+        --from-literal=ARK_SUBDOMAIN=$ARK_SUBDOMAIN
+
+helm upgrade agent "oci://${OCI_BASE}/charts/venafi-kubernetes-agent" \
+     --install \
+     --create-namespace \
+     --namespace "$k8s_namespace" \
+     --version "${RELEASE_HELM_CHART_VERSION}" \
+     --set fullnameOverride=agent \
+     --set "image.repository=${OCI_BASE}/images/venafi-agent" \
+     --values "${script_dir}/values.agent.yaml"
+
+kubectl scale -n "$k8s_namespace" deployment agent  --replicas=0
+kubectl get cm -n "$k8s_namespace" agent-config -o jsonpath={.data.config\\.yaml} > config.original.yaml
+yq eval-all '. as $item ireduce ({}; . * $item)' config.original.yaml "${script_dir}/config.yaml" > config.yaml
+kubectl delete cm -n "$k8s_namespace" agent-config
+kubectl create cm -n "$k8s_namespace" agent-config --from-file=config.yaml
+kubectl scale -n "$k8s_namespace" deployment agent  --replicas=1
diff --git a/hack/e2e/ca/values.agent.yaml b/hack/e2e/ca/values.agent.yaml
new file mode 100644
index 00000000..b7db2541
--- /dev/null
+++ b/hack/e2e/ca/values.agent.yaml
@@ -0,0 +1 @@
+# Empty
diff --git a/pkg/agent/config.go b/pkg/agent/config.go
index 2686a1bd..55ac5352 100644
--- a/pkg/agent/config.go
+++ b/pkg/agent/config.go
@@ -606,12 +606,12 @@ func ValidateAndCombineConfig(log logr.Logger, cfg Config, flags AgentCmdFlags)
 		res.ClusterID = clusterID
 		res.ClusterDescription = cfg.ClusterDescription
 
-		// Validation of `data-gatherers`.
-		if dgErr := ValidateDataGatherers(cfg.DataGatherers); dgErr != nil {
-			errs = multierror.Append(errs, dgErr)
-		}
-		res.DataGatherers = cfg.DataGatherers
 	}
+	// Validation of `data-gatherers`.
+	if dgErr := ValidateDataGatherers(cfg.DataGatherers); dgErr != nil {
+		errs = multierror.Append(errs, dgErr)
+	}
+	res.DataGatherers = cfg.DataGatherers
 
 	// Validation of --period, -p, and the `period` field, as well as
 	// --backoff-max-time, --one-shot, and --strict. The flag --period/-p takes
diff --git a/pkg/agent/run.go b/pkg/agent/run.go
index 6995c548..4beaeb96 100644
--- a/pkg/agent/run.go
+++ b/pkg/agent/run.go
@@ -32,9 +32,11 @@ import (
 
 	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/client"
-	"github.com/jetstack/preflight/pkg/clusteruid"
 	"github.com/jetstack/preflight/pkg/datagatherer"
 	"github.com/jetstack/preflight/pkg/datagatherer/k8s"
+	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
+	"github.com/jetstack/preflight/pkg/internal/cyberark/identity"
+	"github.com/jetstack/preflight/pkg/internal/cyberark/servicediscovery"
 	"github.com/jetstack/preflight/pkg/kubeconfig"
 	"github.com/jetstack/preflight/pkg/logs"
 	"github.com/jetstack/preflight/pkg/version"
@@ -79,26 +81,42 @@ func Run(cmd *cobra.Command, args []string) (returnErr error) {
 		return fmt.Errorf("While evaluating configuration: %v", err)
 	}
 
-	// We need the cluster UID before we progress further so it can be sent along with other data readings
-
-	{
-		restCfg, err := kubeconfig.LoadRESTConfig("")
-		if err != nil {
-			return err
+	var caClient *dataupload.CyberArkClient
+	if config.MachineHubMode {
+		platformDomain := os.Getenv("ARK_PLATFORM_DOMAIN")
+		subdomain := os.Getenv("ARK_SUBDOMAIN")
+		username := os.Getenv("ARK_USERNAME")
+		password := []byte(os.Getenv("ARK_SECRET"))
+
+		const (
+			discoveryContextServiceName = "inventory"
+			separator                   = "."
+		)
+
+		// TODO(wallrj): Maybe get this URL via the service discovery API.
+		// https://platform-discovery.integration-cyberark.cloud/api/public/tenant-discovery?allEndpoints=true&bySubdomain=tlskp-test
+		serviceURL := fmt.Sprintf("https://%s%s%s.%s", subdomain, separator, discoveryContextServiceName, platformDomain)
+
+		var (
+			identityClient *identity.Client
+			err            error
+		)
+		if platformDomain == "cyberark.cloud" {
+			identityClient, err = identity.New(ctx, subdomain)
+		} else {
+			discoveryClient := servicediscovery.New(servicediscovery.WithIntegrationEndpoint())
+			identityClient, err = identity.NewWithDiscoveryClient(ctx, discoveryClient, subdomain)
 		}
-
-		clientset, err := kubernetes.NewForConfig(restCfg)
 		if err != nil {
-			return err
+			return fmt.Errorf("while creating the CyberArk identity client: %v", err)
 		}
-
-		ctx, err = clusteruid.GetClusterUID(ctx, clientset)
+		if err := identityClient.LoginUsernamePassword(ctx, username, password); err != nil {
+			return fmt.Errorf("while logging in: %v", err)
+		}
+		caClient, err = dataupload.NewCyberArkClient(nil, serviceURL, identityClient.AuthenticateRequest)
 		if err != nil {
-			return fmt.Errorf("failed to get cluster UID: %v", err)
+			return fmt.Errorf("while creating the CyberArk dataupload client: %v", err)
 		}
-
-		clusterUID := clusteruid.ClusterUIDFromContext(ctx)
-		log.V(logs.Debug).Info("Retrieved cluster UID", "clusterUID", clusterUID)
 	}
 
 	group, gctx := errgroup.WithContext(ctx)
@@ -262,7 +280,7 @@ func Run(cmd *cobra.Command, args []string) (returnErr error) {
 	// be cancelled, which will cause this blocking loop to exit
 	// instead of waiting for the time period.
 	for {
-		if err := gatherAndOutputData(klog.NewContext(ctx, log), eventf, config, preflightClient, dataGatherers); err != nil {
+		if err := gatherAndOutputData(klog.NewContext(ctx, log), eventf, config, preflightClient, caClient, dataGatherers); err != nil {
 			return err
 		}
 
@@ -316,7 +334,7 @@ func newEventf(log logr.Logger, installNS string) (Eventf, error) {
 // Like Printf but for sending events to the agent's Pod object.
 type Eventf func(eventType, reason, msg string, args ...interface{})
 
-func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConfig, preflightClient client.Client, dataGatherers map[string]datagatherer.DataGatherer) error {
+func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConfig, preflightClient client.Client, caClient *dataupload.CyberArkClient, dataGatherers map[string]datagatherer.DataGatherer) error {
 	log := klog.FromContext(ctx).WithName("gatherAndOutputData")
 	var readings []*api.DataReading
 
@@ -362,8 +380,7 @@ func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConf
 
 		if config.MachineHubMode {
 			post := func() (any, error) {
-				log.Info("machine hub mode not yet implemented")
-				return struct{}{}, nil
+				return struct{}{}, caClient.PostDataReadingsWithOptions(ctx, readings, dataupload.Options{})
 			}
 
 			group.Go(func() error {
diff --git a/pkg/client/client_cyberark.go b/pkg/client/client_cyberark.go
deleted file mode 100644
index af26e96c..00000000
--- a/pkg/client/client_cyberark.go
+++ /dev/null
@@ -1,9 +0,0 @@
-package client
-
-import (
-	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
-)
-
-type CyberArkClient = dataupload.CyberArkClient
-
-var NewCyberArkClient = dataupload.NewCyberArkClient
diff --git a/pkg/clusteruid/clusteruid.go b/pkg/clusteruid/clusteruid.go
deleted file mode 100644
index 2a5327f2..00000000
--- a/pkg/clusteruid/clusteruid.go
+++ /dev/null
@@ -1,45 +0,0 @@
-package clusteruid
-
-import (
-	"context"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-)
-
-// clusterUIDKey is the context key for storing the cluster UID
-type clusterUIDKey struct{}
-
-// GetClusterUID retrieves the UID of the kube-system namespace using the given Kubernetes clientset.
-// This UID can be used as a unique identifier for the Kubernetes cluster.
-// The UID is stored in the given context for later retrieval; use ClusterUIDFromContext to get it.
-func GetClusterUID(ctx context.Context, clientset kubernetes.Interface) (context.Context, error) {
-	namespace, err := clientset.CoreV1().Namespaces().Get(ctx, "kube-system", metav1.GetOptions{})
-	if err != nil {
-		return ctx, err
-	}
-
-	ctx = withClusterUID(ctx, string(namespace.ObjectMeta.UID))
-	return ctx, nil
-}
-
-// ClusterUIDFromContext retrieves the cluster UID from the context.
-// Panics if the value is not found or if the value is not a string.
-func ClusterUIDFromContext(ctx context.Context) string {
-	value := ctx.Value(clusterUIDKey{})
-	if value == nil {
-		panic("cluster UID not found in context")
-	}
-
-	uid, ok := value.(string)
-	if !ok {
-		panic("cluster UID in context is not a string")
-	}
-
-	return uid
-}
-
-// withClusterUID adds the given cluster UID to the context
-func withClusterUID(ctx context.Context, clusterUID string) context.Context {
-	return context.WithValue(ctx, clusterUIDKey{}, clusterUID)
-}
diff --git a/pkg/clusteruid/clusteruid_test.go b/pkg/clusteruid/clusteruid_test.go
deleted file mode 100644
index 1d1cacae..00000000
--- a/pkg/clusteruid/clusteruid_test.go
+++ /dev/null
@@ -1,39 +0,0 @@
-package clusteruid
-
-import (
-	"testing"
-
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/kubernetes/fake"
-)
-
-func TestGetClusterUID(t *testing.T) {
-	client := fake.NewSimpleClientset()
-
-	mockUID := "12345678-1234-5678-1234-567812345678"
-
-	kubeSystemNS := &corev1.Namespace{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: "kube-system",
-			UID:  types.UID(mockUID),
-		},
-	}
-
-	_, err := client.CoreV1().Namespaces().Create(t.Context(), kubeSystemNS, metav1.CreateOptions{})
-	if err != nil {
-		t.Fatalf("failed to create kube-system namespace with fake client: %v", err)
-	}
-
-	ctx, err := GetClusterUID(t.Context(), client)
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	uid := ClusterUIDFromContext(ctx)
-
-	if uid != mockUID {
-		t.Fatalf("expected to get uid=%v, but got uid=%v", mockUID, uid)
-	}
-}
diff --git a/pkg/datagatherer/k8s/discovery.go b/pkg/datagatherer/k8s/discovery.go
index 586622d6..dc913c89 100644
--- a/pkg/datagatherer/k8s/discovery.go
+++ b/pkg/datagatherer/k8s/discovery.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/client-go/discovery"
 
 	"github.com/jetstack/preflight/pkg/datagatherer"
@@ -57,17 +58,18 @@ func (g *DataGathererDiscovery) WaitForCacheSync(ctx context.Context) error {
 	return nil
 }
 
+type DiscoveryData struct {
+	ServerVersion *version.Info `json:"server_version"`
+}
+
 // Fetch will fetch discovery data from the apiserver, or return an error
 func (g *DataGathererDiscovery) Fetch() (interface{}, int, error) {
-	data, err := g.cl.ServerVersion()
+	serverVersion, err := g.cl.ServerVersion()
 	if err != nil {
 		return nil, -1, fmt.Errorf("failed to get server version: %v", err)
 	}
 
-	response := map[string]interface{}{
-		// data has type Info: https://godoc.org/k8s.io/apimachinery/pkg/version#Info
-		"server_version": data,
-	}
-
-	return response, len(response), nil
+	return &DiscoveryData{
+		ServerVersion: serverVersion,
+	}, 1, nil
 }
diff --git a/pkg/datagatherer/k8s/dynamic.go b/pkg/datagatherer/k8s/dynamic.go
index b0e1dedf..aabc3aca 100644
--- a/pkg/datagatherer/k8s/dynamic.go
+++ b/pkg/datagatherer/k8s/dynamic.go
@@ -307,6 +307,10 @@ func (g *DataGathererDynamic) WaitForCacheSync(ctx context.Context) error {
 	return nil
 }
 
+type DynamicData struct {
+	Items []*api.GatheredResource `json:"items"`
+}
+
 // Fetch will fetch the requested data from the apiserver, or return an error
 // if fetching the data fails.
 func (g *DataGathererDynamic) Fetch() (interface{}, int, error) {
@@ -314,7 +318,6 @@ func (g *DataGathererDynamic) Fetch() (interface{}, int, error) {
 		return nil, -1, fmt.Errorf("resource type must be specified")
 	}
 
-	var list = map[string]interface{}{}
 	var items = []*api.GatheredResource{}
 
 	fetchNamespaces := g.namespaces
@@ -344,10 +347,9 @@ func (g *DataGathererDynamic) Fetch() (interface{}, int, error) {
 		return nil, -1, err
 	}
 
-	// add gathered resources to items
-	list["items"] = items
-
-	return list, len(items), nil
+	return &DynamicData{
+		Items: items,
+	}, len(items), nil
 }
 
 func redactList(list []*api.GatheredResource, excludeAnnotKeys, excludeLabelKeys []*regexp.Regexp) error {
diff --git a/pkg/datagatherer/k8s/dynamic_test.go b/pkg/datagatherer/k8s/dynamic_test.go
index 072c4c1c..414b6429 100644
--- a/pkg/datagatherer/k8s/dynamic_test.go
+++ b/pkg/datagatherer/k8s/dynamic_test.go
@@ -730,15 +730,12 @@ func TestDynamicGatherer_Fetch(t *testing.T) {
 			}
 
 			if tc.expected != nil {
-				items, ok := res.(map[string]interface{})
+				data, ok := res.(*DynamicData)
 				if !ok {
-					t.Errorf("expected result be an map[string]interface{} but wasn't")
+					t.Errorf("expected result be *DynamicData but wasn't")
 				}
 
-				list, ok := items["items"].([]*api.GatheredResource)
-				if !ok {
-					t.Errorf("expected result be an []*api.GatheredResource but wasn't")
-				}
+				list := data.Items
 				// sorting list of results by name
 				sortGatheredResources(list)
 				// sorting list of expected results by name
@@ -1045,10 +1042,9 @@ func TestDynamicGathererNativeResources_Fetch(t *testing.T) {
 			}
 
 			if tc.expected != nil {
-				res, ok := rawRes.(map[string]interface{})
-				require.Truef(t, ok, "expected result be an map[string]interface{} but wasn't")
-				actual := res["items"].([]*api.GatheredResource)
-				require.Truef(t, ok, "expected result be an []*api.GatheredResource but wasn't")
+				res, ok := rawRes.(*DynamicData)
+				require.Truef(t, ok, "expected result be an *DynamicData but wasn't")
+				actual := res.Items
 
 				// sorting list of results by name
 				sortGatheredResources(actual)
diff --git a/pkg/datagatherer/k8s/fieldfilter.go b/pkg/datagatherer/k8s/fieldfilter.go
index ed39acb3..1bddd387 100644
--- a/pkg/datagatherer/k8s/fieldfilter.go
+++ b/pkg/datagatherer/k8s/fieldfilter.go
@@ -16,6 +16,9 @@ var SecretSelectedFields = []FieldPath{
 	{"metadata", "ownerReferences"},
 	{"metadata", "selfLink"},
 	{"metadata", "uid"},
+	{"metadata", "creationTimestamp"},
+	{"metadata", "deletionTimestamp"},
+	{"metadata", "resourceVersion"},
 
 	{"type"},
 	{"data", "tls.crt"},
diff --git a/pkg/internal/cyberark/dataupload/dataupload.go b/pkg/internal/cyberark/dataupload/dataupload.go
index a334be2a..75170ec9 100644
--- a/pkg/internal/cyberark/dataupload/dataupload.go
+++ b/pkg/internal/cyberark/dataupload/dataupload.go
@@ -12,9 +12,11 @@ import (
 	"net/http"
 	"net/url"
 
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/transport"
 
 	"github.com/jetstack/preflight/api"
+	"github.com/jetstack/preflight/pkg/datagatherer/k8s"
 	"github.com/jetstack/preflight/pkg/version"
 )
 
@@ -29,6 +31,113 @@ const (
 	apiPathSnapshotLinks = "/api/ingestions/kubernetes/snapshot-links"
 )
 
+type ResourceData map[string][]*unstructured.Unstructured
+
+// Snapshot is the JSON that the CyberArk Discovery and Context API expects to
+// be uploaded to the AWS presigned URL.
+type Snapshot struct {
+	AgentVersion    string                       `json:"agent_version"`
+	ClusterID       string                       `json:"cluster_id"`
+	K8SVersion      string                       `json:"k8s_version"`
+	Secrets         []*unstructured.Unstructured `json:"secrets"`
+	ServiceAccounts []*unstructured.Unstructured `json:"service_accounts"`
+	Roles           []*unstructured.Unstructured `json:"roles"`
+	RoleBindings    []*unstructured.Unstructured `json:"role_bindings"`
+}
+
+// The names of Datagatherers which have the data to populate the Cyberark Snapshot mapped to the key in the Cyberark snapshot.
+var gathererNameToresourceDataKeyMap = map[string]string{
+	"ark/secrets":             "secrets",
+	"ark/serviceaccounts":     "serviceaccounts",
+	"ark/roles":               "roles",
+	"ark/clusterroles":        "roles",
+	"ark/rolebindings":        "rolebindings",
+	"ark/clusterrolebindings": "rolebindings",
+}
+
+func extractResourceListFromReading(reading *api.DataReading) ([]*unstructured.Unstructured, error) {
+	data, ok := reading.Data.(*k8s.DynamicData)
+	if !ok {
+		return nil, fmt.Errorf("failed to convert data: %s", reading.DataGatherer)
+	}
+	items := data.Items
+	resources := make([]*unstructured.Unstructured, len(items))
+	for i, item := range items {
+		if resource, ok := item.Resource.(*unstructured.Unstructured); ok {
+			resources[i] = resource
+		} else {
+			return nil, fmt.Errorf("failed to convert resource: %#v", item)
+		}
+	}
+	return resources, nil
+}
+
+func extractClusterUIDFromReading(reading *api.DataReading) (string, error) {
+	resources, err := extractResourceListFromReading(reading)
+	if err != nil {
+		return "", err
+	}
+	for _, resource := range resources {
+		if resource.GetName() == "kube-system" {
+			return string(resource.GetUID()), nil
+		}
+	}
+	return "", fmt.Errorf("kube-system namespace UID not found in data reading: %v", reading)
+}
+
+func extractServerVersionFromReading(reading *api.DataReading) (string, error) {
+	data, ok := reading.Data.(*k8s.DiscoveryData)
+	if !ok {
+		return "", fmt.Errorf("failed to convert data: %s", reading.DataGatherer)
+	}
+	if data.ServerVersion == nil {
+		return "unknown", nil
+	}
+	return data.ServerVersion.GitVersion, nil
+}
+
+// ConvertDataReadingsToCyberarkSnapshot converts jetstack-secure DataReadings into Cyberark Snapshot format.
+func ConvertDataReadingsToCyberarkSnapshot(
+	readings []*api.DataReading,
+) (_ *Snapshot, err error) {
+	k8sVersion := ""
+	// TODO(wallrj): Use the k8s-discovery gatherer to get clusterID
+	clusterID := ""
+	resourceData := ResourceData{}
+	for _, reading := range readings {
+		if reading.DataGatherer == "ark/discovery" {
+			k8sVersion, err = extractServerVersionFromReading(reading)
+			if err != nil {
+				return nil, fmt.Errorf("while extracting server version from data-reading: %s", err)
+			}
+		}
+		if reading.DataGatherer == "ark/namespaces" {
+			clusterID, err = extractClusterUIDFromReading(reading)
+			if err != nil {
+				return nil, fmt.Errorf("while extracting cluster UID from data-reading: %s", err)
+			}
+		}
+		if key, found := gathererNameToresourceDataKeyMap[reading.DataGatherer]; found {
+			var resources []*unstructured.Unstructured
+			resources, err = extractResourceListFromReading(reading)
+			if err != nil {
+				return nil, fmt.Errorf("while extracting resource list from data-reading: %s", err)
+			}
+			resourceData[key] = append(resourceData[key], resources...)
+		}
+	}
+
+	return &Snapshot{
+		AgentVersion:    version.PreflightVersion,
+		K8SVersion:      k8sVersion,
+		ClusterID:       clusterID,
+		Secrets:         resourceData["secrets"],
+		ServiceAccounts: resourceData["serviceaccounts"],
+		Roles:           resourceData["roles"],
+		RoleBindings:    resourceData["rolebindings"],
+	}, nil
+}
+
 type CyberArkClient struct {
 	baseURL string
 	client  *http.Client
@@ -58,14 +167,19 @@ func NewCyberArkClient(trustedCAs *x509.CertPool, baseURL string, authenticateRe
 // PostDataReadingsWithOptions PUTs the supplied payload to an [AWS presigned URL] which it obtains via the CyberArk inventory API.
 //
 // [AWS presigned URL]: https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html
-func (c *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, payload api.DataReadingsPost, opts Options) error {
+func (c *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, readings []*api.DataReading, opts Options) error {
 	if opts.ClusterName == "" {
 		return fmt.Errorf("programmer mistake: the cluster name (aka `cluster_id` in the config file) cannot be left empty")
 	}
 
+	snapshot, err := ConvertDataReadingsToCyberarkSnapshot(readings)
+	if err != nil {
+		return fmt.Errorf("while converting datareadings to Cyberark snapshot format: %s", err)
+	}
+
 	encodedBody := &bytes.Buffer{}
 	checksum := sha256.New()
-	if err := json.NewEncoder(io.MultiWriter(encodedBody, checksum)).Encode(payload); err != nil {
+	if err := json.NewEncoder(io.MultiWriter(encodedBody, checksum)).Encode(snapshot); err != nil {
 		return err
 	}
 
diff --git a/pkg/internal/cyberark/dataupload/dataupload_test.go b/pkg/internal/cyberark/dataupload/dataupload_test.go
index cadb296a..29d01dbe 100644
--- a/pkg/internal/cyberark/dataupload/dataupload_test.go
+++ b/pkg/internal/cyberark/dataupload/dataupload_test.go
@@ -1,19 +1,27 @@
 package dataupload_test
 
 import (
+	"bytes"
+	"compress/gzip"
 	"crypto/x509"
+	"encoding/json"
 	"encoding/pem"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"k8s.io/klog/v2"
 	"k8s.io/klog/v2/ktesting"
 
 	"github.com/jetstack/preflight/api"
+	"github.com/jetstack/preflight/pkg/datagatherer/k8s"
 	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
 	"github.com/jetstack/preflight/pkg/internal/cyberark/identity"
 	"github.com/jetstack/preflight/pkg/internal/cyberark/servicediscovery"
@@ -21,22 +29,15 @@ import (
 	_ "k8s.io/klog/v2/ktesting/init"
 )
 
-func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
+func TestCyberArkClient_PostDataReadingsWithOptions_MockAPI(t *testing.T) {
 	fakeTime := time.Unix(123, 0)
-	defaultPayload := api.DataReadingsPost{
-		AgentMetadata: &api.AgentMetadata{
-			Version:   "test-version",
-			ClusterID: "test",
-		},
-		DataGatherTime: fakeTime,
-		DataReadings: []*api.DataReading{
-			{
-				ClusterID:     "success-cluster-id",
-				DataGatherer:  "test-gatherer",
-				Timestamp:     api.Time{Time: fakeTime},
-				Data:          map[string]interface{}{"test": "data"},
-				SchemaVersion: "v1",
-			},
+	defaultPayload := []*api.DataReading{
+		{
+			ClusterID:     "success-cluster-id",
+			DataGatherer:  "test-gatherer",
+			Timestamp:     api.Time{Time: fakeTime},
+			Data:          map[string]interface{}{"test": "data"},
+			SchemaVersion: "v1",
 		},
 	}
 	defaultOpts := dataupload.Options{
@@ -52,7 +53,7 @@ func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
 
 	tests := []struct {
 		name         string
-		payload      api.DataReadingsPost
+		payload      []*api.DataReading
 		authenticate func(req *http.Request) error
 		opts         dataupload.Options
 		requireFn    func(t *testing.T, err error)
@@ -84,6 +85,17 @@ func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
 				require.ErrorContains(t, err, "while retrieving snapshot upload URL: received response with status code 500: should authenticate using the correct bearer token")
 			},
 		},
+		{
+			name:    "error contains authenticate error",
+			payload: defaultPayload,
+			opts:    defaultOpts,
+			authenticate: func(_ *http.Request) error {
+				return errors.New("simulated-authenticate-error")
+			},
+			requireFn: func(t *testing.T, err error) {
+				require.ErrorContains(t, err, "while retrieving snapshot upload URL: failed to authenticate request: simulated-authenticate-error")
+			},
+		},
 		{
 			name:         "invalid JSON from server (RetrievePresignedUploadURL step)",
 			payload:      defaultPayload,
@@ -106,6 +118,9 @@ func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
+			logger := ktesting.NewLogger(t, ktesting.DefaultConfig)
+			ctx := klog.NewContext(t.Context(), logger)
+
 			server := dataupload.MockDataUploadServer()
 			defer server.Close()
 
@@ -118,7 +133,7 @@ func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
 			cyberArkClient, err := dataupload.NewCyberArkClient(certPool, server.Server.URL, tc.authenticate)
 			require.NoError(t, err)
 
-			err = cyberArkClient.PostDataReadingsWithOptions(t.Context(), tc.payload, tc.opts)
+			err = cyberArkClient.PostDataReadingsWithOptions(ctx, tc.payload, tc.opts)
 			tc.requireFn(t, err)
 		})
 	}
@@ -132,8 +147,8 @@ func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
 // To enable verbose request logging:
 //
 //	go test ./pkg/internal/cyberark/dataupload/... \
-//	  -v -count 1 -run TestPostDataReadingsWithOptionsWithRealAPI -args -testing.v 6
-func TestPostDataReadingsWithOptionsWithRealAPI(t *testing.T) {
+//	  -v -count 1 -run TestCyberArkClient_PostDataReadingsWithOptions_RealAPI -args -testing.v 6
+func TestCyberArkClient_PostDataReadingsWithOptions_RealAPI(t *testing.T) {
 	platformDomain := os.Getenv("ARK_PLATFORM_DOMAIN")
 	subdomain := os.Getenv("ARK_SUBDOMAIN")
 	username := os.Getenv("ARK_USERNAME")
@@ -172,8 +187,90 @@ func TestPostDataReadingsWithOptionsWithRealAPI(t *testing.T) {
 	cyberArkClient, err := dataupload.NewCyberArkClient(nil, serviceURL, identityClient.AuthenticateRequest)
 	require.NoError(t, err)
 
-	err = cyberArkClient.PostDataReadingsWithOptions(ctx, api.DataReadingsPost{}, dataupload.Options{
-		ClusterName: "bb068932-c80d-460d-88df-34bc7f3f3297",
-	})
+	dataReadings := parseDataReadings(t, readGZIP(t, "testdata/example-1/datareadings.json.gz"))
+	err = cyberArkClient.PostDataReadingsWithOptions(
+		ctx,
+		dataReadings,
+		dataupload.Options{
+			ClusterName: "bb068932-c80d-460d-88df-34bc7f3f3297",
+		},
+	)
 	require.NoError(t, err)
 }
+
+func parseDataReadings(t *testing.T, data []byte) []*api.DataReading {
+	var dataReadings []*api.DataReading
+
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	decoder.DisallowUnknownFields()
+	err := decoder.Decode(&dataReadings)
+	require.NoError(t, err)
+
+	for _, reading := range dataReadings {
+		dataBytes, err := json.Marshal(reading.Data)
+		require.NoError(t, err)
+		in := bytes.NewReader(dataBytes)
+		d := json.NewDecoder(in)
+		d.DisallowUnknownFields()
+
+		var dynamicGatherData k8s.DynamicData
+		if err := d.Decode(&dynamicGatherData); err == nil {
+			reading.Data = &dynamicGatherData
+			continue
+		}
+
+		_, err = in.Seek(0, 0)
+		require.NoError(t, err)
+
+		var discoveryData k8s.DiscoveryData
+		if err = d.Decode(&discoveryData); err == nil {
+			reading.Data = &discoveryData
+			continue
+		}
+
+		require.Failf(t, "failed to parse reading", "reading: %#v", reading)
+	}
+	return dataReadings
+}
+
+func readGZIP(t *testing.T, path string) []byte {
+	f, err := os.Open(path)
+	require.NoError(t, err)
+	defer func() { require.NoError(t, f.Close()) }()
+	gzr, err := gzip.NewReader(f)
+	require.NoError(t, err)
+	defer func() { require.NoError(t, gzr.Close()) }()
+	bytes, err := io.ReadAll(gzr)
+	require.NoError(t, err)
+	return bytes
+}
+
+func writeGZIP(t *testing.T, path string, data []byte) {
+	tmp, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".*")
+	require.NoError(t, err)
+	gzw := gzip.NewWriter(tmp)
+	_, err = io.Copy(gzw, bytes.NewReader(data))
+	require.NoError(t, gzw.Flush())
+	require.NoError(t, gzw.Close())
+	require.NoError(t, tmp.Close())
+	require.NoError(t, err)
+	err = os.Rename(tmp.Name(), path)
+	require.NoError(t, err)
+}
+
+func TestConvertDataReadingsToCyberarkSnapshot(t *testing.T) {
+	dataReadings := parseDataReadings(t, readGZIP(t, "testdata/example-1/datareadings.json.gz"))
+	snapshot, err := dataupload.ConvertDataReadingsToCyberarkSnapshot(dataReadings)
+	require.NoError(t, err)
+
+	actualSnapshotBytes, err := json.MarshalIndent(snapshot, "", "  ")
+	require.NoError(t, err)
+
+	goldenFilePath := "testdata/example-1/snapshot.json.gz"
+	if _, update := os.LookupEnv("UPDATE_GOLDEN_FILES"); update {
+		writeGZIP(t, goldenFilePath, actualSnapshotBytes)
+	} else {
+		expectedSnapshotBytes := readGZIP(t, goldenFilePath)
+		assert.JSONEq(t, string(expectedSnapshotBytes), string(actualSnapshotBytes))
+	}
+}
diff --git a/pkg/internal/cyberark/dataupload/mock.go b/pkg/internal/cyberark/dataupload/mock.go
index f54f0fd3..9e8d1c0d 100644
--- a/pkg/internal/cyberark/dataupload/mock.go
+++ b/pkg/internal/cyberark/dataupload/mock.go
@@ -1,6 +1,7 @@
 package dataupload
 
 import (
+	"bytes"
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
@@ -41,10 +42,7 @@ func (mds *mockDataUploadServer) ServeHTTP(w http.ResponseWriter, r *http.Reques
 		mds.handlePresignedUpload(w, r)
 		return
 	case "/presigned-upload":
-		mds.handleUpload(w, r, false)
-		return
-	case "/presigned-upload-invalid-json":
-		mds.handleUpload(w, r, false)
+		mds.handleUpload(w, r)
 		return
 	default:
 		w.WriteHeader(http.StatusNotFound)
@@ -54,7 +52,10 @@ func (mds *mockDataUploadServer) ServeHTTP(w http.ResponseWriter, r *http.Reques
 func (mds *mockDataUploadServer) handlePresignedUpload(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		w.WriteHeader(http.StatusMethodNotAllowed)
-		_, _ = w.Write([]byte(`{"message":"method not allowed"}`))
+		_, err := w.Write([]byte(`{"message":"method not allowed"}`))
+		if err != nil {
+			panic(err)
+		}
 		return
 	}
 
@@ -98,7 +99,10 @@ func (mds *mockDataUploadServer) handlePresignedUpload(w http.ResponseWriter, r
 	if req.ClusterID == "invalid-json-retrieve-presigned" {
 		w.WriteHeader(http.StatusOK)
 		w.Header().Set("Content-Type", "application/json")
-		_, _ = w.Write([]byte(`{"url":`)) // invalid JSON
+		_, err := w.Write([]byte(`{"url":`)) // invalid JSON
+		if err != nil {
+			panic(err)
+		}
 		return
 	}
 
@@ -122,10 +126,13 @@ func (mds *mockDataUploadServer) handlePresignedUpload(w http.ResponseWriter, r
 	}{presignedURL})
 }
 
-func (mds *mockDataUploadServer) handleUpload(w http.ResponseWriter, r *http.Request, invalidJSON bool) {
+func (mds *mockDataUploadServer) handleUpload(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPut {
 		w.WriteHeader(http.StatusMethodNotAllowed)
-		_, _ = w.Write([]byte(`{"message":"method not allowed"}`))
+		_, err := w.Write([]byte(`{"message":"method not allowed"}`))
+		if err != nil {
+			panic(err)
+		}
 		return
 	}
 
@@ -134,21 +141,26 @@ func (mds *mockDataUploadServer) handleUpload(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	if invalidJSON {
-		w.WriteHeader(http.StatusOK)
-		w.Header().Set("Content-Type", "application/json")
-		_, _ = w.Write([]byte(`{"url":`)) // invalid JSON
-		return
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		panic(err)
 	}
 
 	checksum := sha256.New()
-	_, _ = io.Copy(checksum, r.Body)
-
+	_, err = checksum.Write(body)
+	if err != nil {
+		panic(err)
+	}
 	if r.URL.Query().Get("checksum") != hex.EncodeToString(checksum.Sum(nil)) {
 		http.Error(w, "checksum is invalid", http.StatusInternalServerError)
 	}
 
+	var snapshot Snapshot
+	d := json.NewDecoder(bytes.NewBuffer(body))
+	d.DisallowUnknownFields()
+	if err := d.Decode(&snapshot); err != nil {
+		panic(err)
+	}
+
 	w.WriteHeader(http.StatusOK)
-	w.Header().Set("Content-Type", "application/json")
-	_, _ = w.Write([]byte(`{"success":true}`))
 }
diff --git a/pkg/internal/cyberark/dataupload/testdata/example-1/README.md b/pkg/internal/cyberark/dataupload/testdata/example-1/README.md
new file mode 100644
index 00000000..0ea38f86
--- /dev/null
+++ b/pkg/internal/cyberark/dataupload/testdata/example-1/README.md
@@ -0,0 +1,27 @@
+# README
+
+Data captured from a cert-manager E2E test cluster.
+
+```bash
+cd cert-manager
+make e2e-setup
+```
+
+```bash
+cd jetstack-secure
+go run . agent \
+    --api-token not-used \
+    --install-namespace venafi \
+    --log-level 6 \
+    --one-shot \
+    --agent-config-file pkg/internal/cyberark/dataupload/testdata/example-1/agent.yaml \
+    --output-path pkg/internal/cyberark/dataupload/testdata/example-1/datareadings.json
+gzip pkg/internal/cyberark/dataupload/testdata/example-1/datareadings.json
+```
+
+
+To recreate the golden output file:
+
+```bash
+UPDATE_GOLDEN_FILES=true go test ./pkg/internal/cyberark/dataupload/... -run TestConvertDataReadingsToCyberarkSnapshot
+```
diff --git a/pkg/internal/cyberark/dataupload/testdata/example-1/agent.yaml b/pkg/internal/cyberark/dataupload/testdata/example-1/agent.yaml
new file mode 100644
index 00000000..44bf3ef7
--- /dev/null
+++ b/pkg/internal/cyberark/dataupload/testdata/example-1/agent.yaml
@@ -0,0 +1,60 @@
+cluster_id: example-cluster-id
+organization_id: example-organization-id
+data-gatherers:
+# gather k8s apiserver version information
+- kind: k8s-discovery
+  name: ark/discovery
+- kind: k8s-dynamic
+  name: ark/namespaces
+  config:
+    resource-type:
+      version: v1
+      resource: namespaces
+- kind: k8s-dynamic
+  name: ark/serviceaccounts
+  config:
+    resource-type:
+      resource: serviceaccounts
+      version: v1
+- kind: k8s-dynamic
+  name: ark/secrets
+  config:
+    resource-type:
+      version: v1
+      resource: secrets
+    field-selectors:
+    - type!=kubernetes.io/service-account-token
+    - type!=kubernetes.io/dockercfg
+    - type!=kubernetes.io/dockerconfigjson
+    - type!=kubernetes.io/basic-auth
+    - type!=kubernetes.io/ssh-auth
+    - type!=bootstrap.kubernetes.io/token
+    - type!=helm.sh/release.v1
+- kind: k8s-dynamic
+  name: ark/roles
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: roles
+- kind: k8s-dynamic
+  name: ark/clusterroles
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: clusterroles
+- kind: k8s-dynamic
+  name: ark/rolebindings
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: rolebindings
+- kind: k8s-dynamic
+  name: ark/clusterrolebindings
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: clusterrolebindings
diff --git a/pkg/internal/cyberark/dataupload/testdata/example-1/datareadings.json.gz b/pkg/internal/cyberark/dataupload/testdata/example-1/datareadings.json.gz
new file mode 100644
index 00000000..ac34d911
Binary files /dev/null and b/pkg/internal/cyberark/dataupload/testdata/example-1/datareadings.json.gz differ
diff --git a/pkg/internal/cyberark/dataupload/testdata/example-1/snapshot.json.gz b/pkg/internal/cyberark/dataupload/testdata/example-1/snapshot.json.gz
new file mode 100644
index 00000000..506a3058
Binary files /dev/null and b/pkg/internal/cyberark/dataupload/testdata/example-1/snapshot.json.gz differ