Add option to copy the OCI layer to storage

Add on optional field to the `OCIRepository.spec.layerSelector` called `operation` that accepts one of the following values: `extract` or `copy`. When the operation is set to `copy`, instead of extracting the compressed layer, the controller copies the compressed blob as it is to storage, thus keeping the original content unaltered.

Signed-off-by: Stefan Prodan <stefan.prodan@gmail.com>

api/v1beta2/ocirepository_types.go

@@ -45,6 +45,12 @@ const (
 	// AzureOCIProvider provides support for OCI authentication using a Azure Service Principal,
 	// Managed Identity or Shared Key.
 	AzureOCIProvider string = "azure"
+
+	// OCILayerExtract defines the operation type for extracting the content from an OCI artifact layer.
+	OCILayerExtract = "extract"
+
+	// OCILayerCopy defines the operation type for copying the content from an OCI artifact layer.
+	OCILayerCopy = "copy"
 )
 
 // OCIRepositorySpec defines the desired state of OCIRepository
@@ -156,6 +162,14 @@ type OCILayerSelector struct {
 	// first layer matching this type is selected.
 	// +optional
 	MediaType string `json:"mediaType,omitempty"`
+
+	// Operation specifies how the selected layer should be processed.
+	// By default, the layer compressed content is extracted to storage.
+	// When the operation is set to 'copy', the layer compressed content
+	// is persisted to storage as it is.
+	// +kubebuilder:validation:Enum=extract;copy
+	// +optional
+	Operation string `json:"operation,omitempty"`
 }
 
 // OCIRepositoryVerification verifies the authenticity of an OCI Artifact
@@ -231,6 +245,15 @@ func (in *OCIRepository) GetLayerMediaType() string {
 	return in.Spec.LayerSelector.MediaType
 }
 
+// GetLayerOperation returns the layer selector operation (defaults to extract).
+func (in *OCIRepository) GetLayerOperation() string {
+	if in.Spec.LayerSelector == nil || in.Spec.LayerSelector.Operation == "" {
+		return OCILayerExtract
+	}
+
+	return in.Spec.LayerSelector.Operation
+}
+
 // +genclient
 // +genclient:Namespaced
 // +kubebuilder:storageversion

config/crd/bases/source.toolkit.fluxcd.io_ocirepositories.yaml

@@ -90,6 +90,15 @@ spec:
                       which should be extracted from the OCI Artifact. The first layer
                       matching this type is selected.
                     type: string
+                  operation:
+                    description: Operation specifies how the selected layer should
+                      be processed. By default, the layer compressed content is extracted
+                      to storage. When the operation is set to 'copy', the layer compressed
+                      content is persisted to storage as it is.
+                    enum:
+                    - extract
+                    - copy
+                    type: string
                 type: object
               provider:
                 default: generic

controllers/ocirepository_controller.go

@@ -22,8 +22,10 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"os"
+	"path/filepath"
 	"sort"
 	"strings"
 	"time"
@@ -499,6 +501,7 @@ func (r *OCIRepositoryReconciler) reconcileSource(ctx context.Context, obj *sour
 			layer = layers[0]
 		}
 
+		// Extract the compressed content from the selected layer
 		blob, err := layer.Compressed()
 		if err != nil {
 			e := serror.NewGeneric(
@@ -509,9 +512,42 @@ func (r *OCIRepositoryReconciler) reconcileSource(ctx context.Context, obj *sour
 			return sreconcile.ResultEmpty, e
 		}
 
-		if _, err = untar.Untar(blob, dir); err != nil {
+		// Persist layer content to storage using the specified operation
+		switch obj.GetLayerOperation() {
+		case sourcev1.OCILayerExtract:
+			if _, err = untar.Untar(blob, dir); err != nil {
+				e := serror.NewGeneric(
+					fmt.Errorf("failed to extract layer contents from artifact: %w", err),
+					sourcev1.OCILayerOperationFailedReason,
+				)
+				conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, e.Err.Error())
+				return sreconcile.ResultEmpty, e
+			}
+		case sourcev1.OCILayerCopy:
+			metadata.Path = fmt.Sprintf("%s.tgz", metadata.Revision)
+			file, err := os.Create(filepath.Join(dir, metadata.Path))
+			if err != nil {
+				e := serror.NewGeneric(
+					fmt.Errorf("failed to create file to copy layer to: %w", err),
+					sourcev1.OCILayerOperationFailedReason,
+				)
+				conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, e.Err.Error())
+				return sreconcile.ResultEmpty, e
+			}
+			defer file.Close()
+
+			_, err = io.Copy(file, blob)
+			if err != nil {
+				e := serror.NewGeneric(
+					fmt.Errorf("failed to copy layer from artifact: %w", err),
+					sourcev1.OCILayerOperationFailedReason,
+				)
+				conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, e.Err.Error())
+				return sreconcile.ResultEmpty, e
+			}
+		default:
 			e := serror.NewGeneric(
-				fmt.Errorf("failed to untar the first layer from artifact: %w", err),
+				fmt.Errorf("unsupported layer operation: %s", obj.GetLayerOperation()),
 				sourcev1.OCILayerOperationFailedReason,
 			)
 			conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, e.Err.Error())
@@ -915,14 +951,25 @@ func (r *OCIRepositoryReconciler) reconcileArtifact(ctx context.Context,
 	}
 	defer unlock()
 
-	// Archive directory to storage
-	if err := r.Storage.Archive(&artifact, dir, nil); err != nil {
-		e := serror.NewGeneric(
-			fmt.Errorf("unable to archive artifact to storage: %s", err),
-			sourcev1.ArchiveOperationFailedReason,
-		)
-		conditions.MarkTrue(obj, sourcev1.StorageOperationFailedCondition, e.Reason, e.Err.Error())
-		return sreconcile.ResultEmpty, e
+	switch obj.GetLayerOperation() {
+	case sourcev1.OCILayerCopy:
+		if err = r.Storage.CopyFromPath(&artifact, filepath.Join(dir, metadata.Path)); err != nil {
+			e := serror.NewGeneric(
+				fmt.Errorf("unable to copy artifact to storage: %w", err),
+				sourcev1.ArchiveOperationFailedReason,
+			)
+			conditions.MarkTrue(obj, sourcev1.StorageOperationFailedCondition, e.Reason, e.Err.Error())
+			return sreconcile.ResultEmpty, e
+		}
+	default:
+		if err := r.Storage.Archive(&artifact, dir, nil); err != nil {
+			e := serror.NewGeneric(
+				fmt.Errorf("unable to archive artifact to storage: %s", err),
+				sourcev1.ArchiveOperationFailedReason,
+			)
+			conditions.MarkTrue(obj, sourcev1.StorageOperationFailedCondition, e.Reason, e.Err.Error())
+			return sreconcile.ResultEmpty, e
+		}
 	}
 
 	// Record it on the object

controllers/ocirepository_controller_test.go

@@ -85,6 +85,7 @@ func TestOCIRepository_Reconcile(t *testing.T) {
 		semver         string
 		digest         string
 		mediaType      string
+		operation      string
 		assertArtifact []artifactFixture
 	}{
 		{
@@ -93,6 +94,7 @@ func TestOCIRepository_Reconcile(t *testing.T) {
 			tag:       podinfoVersions["6.1.6"].tag,
 			digest:    podinfoVersions["6.1.6"].digest.Hex,
 			mediaType: "application/vnd.docker.image.rootfs.diff.tar.gzip",
+			operation: sourcev1.OCILayerCopy,
 			assertArtifact: []artifactFixture{
 				{
 					expectedPath:     "kustomize/deployment.yaml",
@@ -150,7 +152,12 @@ func TestOCIRepository_Reconcile(t *testing.T) {
 			}
 			if tt.mediaType != "" {
 				obj.Spec.LayerSelector = &sourcev1.OCILayerSelector{MediaType: tt.mediaType}
+
+				if tt.operation != "" {
+					obj.Spec.LayerSelector.Operation = tt.operation
+				}
 			}
+
 			g.Expect(testEnv.Create(ctx, obj)).To(Succeed())
 
 			key := client.ObjectKey{Name: obj.Name, Namespace: obj.Namespace}

docs/api/source.md

@@ -2635,6 +2635,21 @@ which should be extracted from the OCI Artifact. The
 first layer matching this type is selected.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>operation</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Operation specifies how the selected layer should be processed.
+By default, the layer compressed content is extracted to storage.
+When the operation is set to &lsquo;copy&rsquo;, the layer compressed content
+is persisted to storage as it is.</p>
+</td>
+</tr>
 </tbody>
 </table>
 </div>