ONTAP: cache templates on primary storage and clone volumes from them

Enable the managed-storage template cache flow for the ONTAP primary
storage plugin so that VM root volumes are cloned from a template copy
kept on the pool's FlexVolume instead of being copied from secondary
storage for every deployment.

- Advertise CAN_CREATE_VOLUME_FROM_VOLUME from the datastore driver.
- Create the template cache on primary: an iSCSI LUN (identity tracked on
  template_spool_ref) or a qcow2 file for NFS, populated by the KVM host.
- Clone new volumes from the cached template: LUN clone for iSCSI and
  ONTAP file clone (FileCloneRequest) for NFS, growing the clone when the
  disk offering is larger than the template.
- Clean up the cached LUN/file when the template is removed from the pool;
  a missing file is treated as already deleted.
- Move protocol-specific create/clone/resize logic into
  UnifiedSANStrategy and UnifiedNASStrategy, and add a StorageStrategy
  helper to poll an optional async job.
- Add unit tests for the driver and the NAS/SAN strategies.

Author: rajiv-jain-netapp

plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/driver/OntapPrimaryDatastoreDriver.java

@@ -21,7 +21,9 @@
 
 import feign.QueryMap;
 import org.apache.cloudstack.storage.feign.model.ExportPolicy;
+import org.apache.cloudstack.storage.feign.model.FileCloneRequest;
 import org.apache.cloudstack.storage.feign.model.FileInfo;
+import org.apache.cloudstack.storage.feign.model.response.JobResponse;
 import org.apache.cloudstack.storage.feign.model.response.OntapResponse;
 import feign.Headers;
 import feign.Param;
@@ -58,6 +60,15 @@ void createFile(@Param("authHeader") String authHeader,
                     @Param("path") String filePath,
                     FileInfo file);
 
+    /**
+     * Creates a space-efficient clone of a file within a FlexVolume.
+     *
+     * <p>ONTAP REST: {@code POST /api/storage/file/clone}</p>
+     */
+    @RequestLine("POST /api/storage/file/clone")
+    @Headers({"Authorization: {authHeader}", "Content-Type: application/json"})
+    JobResponse cloneFile(@Param("authHeader") String authHeader, FileCloneRequest request);
+
     // Export Policy Operations
     @RequestLine("POST /api/protocols/nfs/export-policies")
     @Headers({"Authorization: {authHeader}"})

plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/feign/client/NASFeignClient.java

@@ -50,8 +50,8 @@ public interface SANFeignClient {
     @Headers({"Authorization: {authHeader}"})
     Lun getLunByUUID(@Param("authHeader") String authHeader, @Param("uuid") String uuid);
 
-    @RequestLine("PATCH /{uuid}")
-    @Headers({"Authorization: {authHeader}"})
+    @RequestLine("PATCH /api/storage/luns/{uuid}")
+    @Headers({"Authorization: {authHeader}", "Content-Type: application/json"})
     void updateLun(@Param("authHeader") String authHeader, @Param("uuid") String uuid, Lun lun);
 
     @RequestLine("DELETE /api/storage/luns/{uuid}")

plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/feign/client/SANFeignClient.java

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.cloudstack.storage.feign.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Request body for the ONTAP file clone API.
+ *
+ * <p>ONTAP REST endpoint: {@code POST /api/storage/file/clone}</p>
+ *
+ * <p>Creates a space-efficient copy of a file. Source and destination paths are relative to the
+ * root of {@code volume}, and both must live in that same FlexVolume.</p>
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FileCloneRequest {
+
+    @JsonProperty("volume")
+    private VolumeRef volume;
+
+    @JsonProperty("source_path")
+    private String sourcePath;
+
+    @JsonProperty("destination_path")
+    private String destinationPath;
+
+    @JsonProperty("overwrite_destination")
+    private Boolean overwriteDestination;
+
+    public FileCloneRequest() {
+    }
+
+    public FileCloneRequest(String flexVolUuid, String flexVolName, String sourcePath, String destinationPath) {
+        this.volume = new VolumeRef(flexVolUuid, flexVolName);
+        this.sourcePath = sourcePath;
+        this.destinationPath = destinationPath;
+    }
+
+    public VolumeRef getVolume() {
+        return volume;
+    }
+
+    public void setVolume(VolumeRef volume) {
+        this.volume = volume;
+    }
+
+    public String getSourcePath() {
+        return sourcePath;
+    }
+
+    public void setSourcePath(String sourcePath) {
+        this.sourcePath = sourcePath;
+    }
+
+    public String getDestinationPath() {
+        return destinationPath;
+    }
+
+    public void setDestinationPath(String destinationPath) {
+        this.destinationPath = destinationPath;
+    }
+
+    public Boolean getOverwriteDestination() {
+        return overwriteDestination;
+    }
+
+    public void setOverwriteDestination(Boolean overwriteDestination) {
+        this.overwriteDestination = overwriteDestination;
+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public static class VolumeRef {
+
+        @JsonProperty("uuid")
+        private String uuid;
+
+        @JsonProperty("name")
+        private String name;
+
+        public VolumeRef() {
+        }
+
+        public VolumeRef(String uuid, String name) {
+            this.uuid = uuid;
+            this.name = name;
+        }
+
+        public String getUuid() {
+            return uuid;
+        }
+
+        public void setUuid(String uuid) {
+            this.uuid = uuid;
+        }
+
+        public String getName() {
+            return name;
+        }
+
+        public void setName(String name) {
+            this.name = name;
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "FileCloneRequest{volume=" + (volume != null ? volume.getUuid() : null)
+                + ", sourcePath=" + sourcePath
+                + ", destinationPath=" + destinationPath + "}";
+    }
+}

plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/feign/model/FileCloneRequest.java

@@ -308,6 +308,7 @@ private String toIndentedString(Object o) {
     }
 
 
+    @JsonInclude(JsonInclude.Include.NON_NULL)
     public static class Clone {
         @JsonProperty("source")
         private Source source = null;
@@ -319,6 +320,7 @@ public void setSource(Source source) {
         }
     }
 
+    @JsonInclude(JsonInclude.Include.NON_NULL)
     public static class Source {
         @JsonProperty("name")
         private String name = null;

plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/feign/model/Lun.java

@@ -40,6 +40,7 @@
 import org.apache.cloudstack.storage.feign.model.Volume;
 import org.apache.cloudstack.storage.feign.model.response.JobResponse;
 import org.apache.cloudstack.storage.feign.model.response.OntapResponse;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.service.model.AccessGroup;
 import org.apache.cloudstack.storage.service.model.CloudStackVolume;
 import org.apache.cloudstack.storage.service.model.ProtocolType;
@@ -54,6 +55,8 @@
 import java.util.Map;
 import java.util.Objects;
 
+import org.apache.cloudstack.engine.subsystem.api.storage.TemplateInfo;
+
 /**
  * Storage Strategy represents the communication path for all the ONTAP storage options
  *
@@ -495,6 +498,27 @@ public String getNetworkInterface() {
      */
     abstract public CloudStackVolume createCloudStackVolume(CloudStackVolume cloudstackVolume);
 
+    /**
+     * Creates the protocol-specific backend object that caches a template on this pool.
+     *
+     * <p>iSCSI creates an empty LUN ({@code /vol/&lt;flexVol&gt;/cs_tmpl_&lt;id&gt;}) sized to
+     * {@code sizeInBytes}. NFS is a no-op on the array: the KVM agent later writes the qcow2
+     * into the mounted FlexVolume.</p>
+     *
+     * <p>Returns a {@link CloudStackVolume} so the driver can map it to {@code CreateCmdResult}
+     * and update {@code template_spool_ref}. SAN populates {@code lun}; NAS returns an empty
+     * volume (no LUN / file yet).</p>
+     *
+     * @param storagePool   CloudStack primary storage pool (one FlexVolume)
+     * @param templateInfo  template being cached
+     * @param details       pool details (SVM, protocol, etc.)
+     * @param sizeInBytes   virtual size for the cache object (required for SAN; ignored for NAS)
+     * @return created cache identity, or an empty {@link CloudStackVolume} when nothing is
+     *         pre-created on the array
+     */
+    abstract public CloudStackVolume createTemplateCache(StoragePoolVO storagePool, TemplateInfo templateInfo,
+            Map<String, String> details, long sizeInBytes);
+
     /**
      * Method encapsulates the behavior based on the opted protocol in subclasses.
      * it is going to mimic
@@ -519,14 +543,28 @@ public String getNetworkInterface() {
     abstract public void deleteCloudStackVolume(CloudStackVolume cloudstackVolume);
 
     /**
-     * Method encapsulates the behavior based on the opted protocol in subclasses.
+     * Creates a space-efficient clone of an existing object inside the same FlexVolume.
      * it is going to mimic
      *     cloneLun       for iSCSI, FC protocols
      *     cloneFile      for NFS3.0 and NFS4.1 protocols
      *     cloneNameSpace for Nvme/TCP and Nvme/FC protocol
-     * @param cloudstackVolume the CloudStack volume to copy
+     *
+     * <p>ONTAP requires the source and the destination to live in the same FlexVolume, which
+     * holds because a CloudStack primary storage pool maps one-to-one onto a FlexVolume.</p>
+     *
+     * @param cloudstackVolume describes the clone to create; the source is carried in the
+     *                         protocol-specific clone reference (for SAN, {@code lun.clone.source})
+     * @return the created CloudStackVolume, populated with the backend identity of the clone
      */
-    abstract public void copyCloudStackVolume(CloudStackVolume cloudstackVolume);
+    abstract public CloudStackVolume cloneCloudStackVolume(CloudStackVolume cloudstackVolume);
+
+    /**
+     * Grows an existing backend object to {@code sizeInBytes}.
+     *
+     * <p>Needed after cloning a cached template, because a clone inherits the size of its source
+     * while the service offering may ask for a larger disk.</p>
+     */
+    abstract public void resizeCloudStackVolume(CloudStackVolume cloudstackVolume, long sizeInBytes);
 
     /**
      * Method encapsulates the behavior based on the opted protocol in subclasses.
@@ -694,4 +732,22 @@ public Boolean jobPollForSuccess(String jobUUID, int maxRetries, int sleepTimeIn
         }
         return true;
     }
+
+    /**
+     * Polls an ONTAP async job when the API response includes a job reference.
+     *
+     * <p>When no job is returned (synchronous completion), the operation is treated as
+     * successful after HTTP 2xx.</p>
+     *
+     * @param response       ONTAP job response (may be null or without a job)
+     * @param operationName  label for logging and error messages
+     */
+    public void pollJobIfPresent(JobResponse response, String operationName) {
+        if (response == null || response.getJob() == null || response.getJob().getUuid() == null) {
+            logger.debug("pollJobIfPresent: No async job returned for operation [{}], continuing without polling",
+                    operationName);
+            return;
+        }
+        jobPollForSuccess(response.getJob().getUuid(), 60, 2000);
+    }
 }