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>