Copilot commented on code in PR #14193:
URL: https://github.com/apache/cloudstack/pull/14193#discussion_r4219669824
##########
plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/service/StorageStrategy.java:
##########
@@ -96,6 +105,139 @@ public StorageStrategy(OntapStorage ontapStorage) {
this.sanFeignClient =
feignClientFactory.createClient(SANFeignClient.class, baseURL);
this.nasFeignClient =
feignClientFactory.createClient(NASFeignClient.class, baseURL);
this.snapshotFeignClient =
feignClientFactory.createClient(SnapshotFeignClient.class, baseURL);
+ this.clusterFeignClient =
feignClientFactory.createClient(ClusterFeignClient.class, baseURL);
+ this.emsFeignClient =
feignClientFactory.createClient(EmsFeignClient.class, baseURL);
+ }
+
+ /**
+ * Fetches the ONTAP {@link Cluster} (name, uuid, version) and, for ASUP,
the rolled-up node
+ * {@code model} and {@code platformType} from {@code GET
/api/cluster/nodes}. Best-effort:
+ * returns {@code null} if the cluster GET fails. Node fields are left
unset if the nodes GET
+ * fails. Callers must never fail a storage operation because of this.
+ *
+ * @return the ONTAP {@link Cluster}, or {@code null} if it cannot be
resolved
+ */
+ public Cluster getClusterInfo() {
+ try {
+ String authHeader =
OntapStorageUtils.generateAuthHeader(storage.getUsername(),
storage.getPassword());
+ Cluster cluster = clusterFeignClient.getCluster(authHeader, true);
+ if (cluster != null) {
+ populateNodeAsupFields(cluster, authHeader);
+ }
+ return cluster;
+ } catch (Exception e) {
+ logger.warn("getClusterInfo: failed to fetch ONTAP cluster info
for storage IP {}: {}",
+ storage.getStorageIP(), e.getMessage());
+ return null;
+ }
+ }
+
+ /**
+ * Sets {@link Cluster#setModel(String)} and {@link
Cluster#setPlatformType(String)} from one
+ * {@code GET /api/cluster/nodes}. Distinct models are joined with a
comma. Platform type is
+ * {@code performance}, {@code capacity}, {@code fas}, or {@code
composite} when personalities mix.
+ */
+ private void populateNodeAsupFields(Cluster cluster, String authHeader) {
+ try {
+ Map<String, Object> query = new HashMap<>();
+ query.put(OntapStorageConstants.FIELDS,
OntapStorageConstants.CLUSTER_NODE_ASUP_FIELDS);
+ OntapResponse<ClusterNode> response =
clusterFeignClient.getClusterNodes(authHeader, query);
+ if (response == null || response.getRecords() == null ||
response.getRecords().isEmpty()) {
+ return;
+ }
+ LinkedHashSet<String> models = new LinkedHashSet<>();
+ LinkedHashSet<String> platformTypes = new LinkedHashSet<>();
+ for (ClusterNode node : response.getRecords()) {
+ if (node == null) {
+ continue;
+ }
+ if (node.getModel() != null && !node.getModel().isBlank()) {
+ models.add(node.getModel().trim());
+ }
+ platformTypes.add(classifyNodePlatformType(node));
+ }
+ if (!models.isEmpty()) {
+ cluster.setModel(String.join(OntapStorageConstants.COMMA,
models));
+ }
+ cluster.setPlatformType(rollupPlatformType(platformTypes));
+ } catch (Exception e) {
+ logger.warn("getClusterInfo: failed to fetch ONTAP node hardware
for storage IP {}: {}",
+ storage.getStorageIP(), e.getMessage());
+ }
+ }
+
+ /**
+ * Extracts a clean, parser-friendly ONTAP version string from a {@link
Cluster}.
+ *
+ * <p>Prefers the compact "generation.major.minor" numeric form (e.g.
"9.17.1"), which avoids
+ * the colon/date noise in {@code version.full}. Falls back to the verbose
{@code version.full}
+ * banner only when the numeric fields are unavailable.</p>
+ *
+ * @param cluster the cluster (may be {@code null})
+ * @return the ONTAP version string, or {@code null} if it cannot be
resolved
+ */
+ public String getClusterVersion(Cluster cluster) {
+ if (cluster == null || cluster.getVersion() == null) {
+ return null;
+ }
+ Version version = cluster.getVersion();
+ if (version.getGeneration() != null && version.getMajor() != null &&
version.getMinor() != null) {
+ return version.getGeneration() + OntapStorageConstants.DOT +
version.getMajor()
+ + OntapStorageConstants.DOT + version.getMinor();
+ }
+ if (version.getFull() != null && !version.getFull().isEmpty()) {
+ return version.getFull();
+ }
+ return null;
+ }
+
+ /**
+ * Classifies one node: not all-flash → {@code fas}; all-flash + capacity
→ {@code capacity};
+ * otherwise all-flash (including performance-optimized or classic AFF) →
{@code performance}.
+ */
+ private static String classifyNodePlatformType(ClusterNode node) {
+ boolean allFlash = Boolean.TRUE.equals(node.getAllFlashOptimized());
+ boolean capacity = Boolean.TRUE.equals(node.getCapacityOptimized());
+ if (!allFlash) {
+ return OntapStorageConstants.ASUP_PLATFORM_TYPE_FAS;
+ }
+ if (capacity) {
+ return OntapStorageConstants.ASUP_PLATFORM_TYPE_CAPACITY;
+ }
+ return OntapStorageConstants.ASUP_PLATFORM_TYPE_PERFORMANCE;
+ }
Review Comment:
This treats a missing/unknown `is_all_flash_optimized` (null) as `false`,
which classifies the node as `fas` and can incorrectly set cluster
`platformType` when ONTAP doesn’t return these fields (permissions/older
versions/partial responses). Consider returning `null` (unknown) when the
required flags are absent and only adding non-null classifications to the
rollup set so ASUP doesn’t misreport hardware type.
##########
plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/asup/OntapAsupManager.java:
##########
@@ -0,0 +1,666 @@
+/*
+ * 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.asup;
+
+import com.cloud.cluster.ManagementServerHostVO;
+import com.cloud.cluster.dao.ManagementServerHostDao;
+import com.cloud.event.EventTypes;
+import com.cloud.server.ManagementService;
+import com.cloud.storage.Volume;
+import com.cloud.storage.SnapshotVO;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.SnapshotDao;
+import com.cloud.storage.dao.VolumeDao;
+import com.cloud.vm.snapshot.VMSnapshot;
+import com.cloud.vm.snapshot.VMSnapshotVO;
+import com.cloud.vm.snapshot.dao.VMSnapshotDao;
+import com.cloud.utils.Ternary;
+import com.cloud.utils.component.ManagerBase;
+import com.cloud.utils.concurrency.NamedThreadFactory;
+import com.cloud.utils.db.GlobalLock;
+import com.cloud.utils.net.NetUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.messagebus.MessageBus;
+import org.apache.cloudstack.managed.context.ManagedContextRunnable;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailsDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
+import org.apache.cloudstack.storage.feign.model.Cluster;
+import org.apache.cloudstack.storage.feign.model.EmsApplicationLog;
+import org.apache.cloudstack.storage.service.StorageStrategy;
+import org.apache.cloudstack.storage.utils.OntapConfigurationManager;
+import org.apache.cloudstack.storage.utils.OntapStorageConstants;
+import org.apache.cloudstack.storage.utils.OntapStorageUtils;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.inject.Inject;
+import javax.naming.ConfigurationException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Periodic ASUP (AutoSupport) telemetry pusher for the NetApp ONTAP plugin.
+ *
+ * <p>Each run pushes, for every ONTAP-backed primary storage pool, two
minimal EMS
+ * application-log messages to the backing ONTAP cluster:</p>
+ *
+ * <ul>
+ * <li><b>event-id 0 (heartbeat):</b> identifies the CloudStack deployment
(CloudStack
+ * version, management host) connected to the ONTAP cluster (ONTAP
cluster version).</li>
+ * <li><b>event-id 1 (pool):</b> maps the CloudStack storage pool to its
backing ONTAP
+ * volume - protocol (NFS/iSCSI), ONTAP FlexVolume UUID, SVM, disk
usage, and
+ * snapshot telemetry (counts by state, total provisioned size).</li>
+ * </ul>
+ *
+ * <p>Runs are booked one at a time: after each cycle starts, the next run is
scheduled for
+ * {@link OntapConfigurationManager#AsupIntervalHours} after that start, so
production
+ * intervals (hours) are start-to-start. REST work sits inside the interval;
it is not added
+ * after it. Editing the ONTAP ASUP interval re-books the pending run
immediately, with no
+ * management-server restart.</p>
+ */
+public class OntapAsupManager extends ManagerBase {
+ private static final int ASUP_LOCK_TIMEOUT_SECONDS = 5;
+
+ /**
+ * Volume states that guarantee a physical object exists on the ONTAP
FlexVolume.
+ * States like {@link Volume.State#Allocated} have a CloudStack DB row
pointing to this
+ * pool but ONTAP provisioning has not been called yet — they must be
excluded to avoid
+ * inflating disk counts and provisioned-size totals. Upload-family states
live on
+ * secondary storage, not on the primary ONTAP volume, so they are also
excluded.
+ */
+ private static final Set<Volume.State> CS_VOLUME_STATES = EnumSet.of(
+ Volume.State.Ready,
+ Volume.State.Snapshotting,
+ Volume.State.RevertSnapshotting,
+ Volume.State.Attaching,
+ Volume.State.Restoring,
+ Volume.State.Expunging,
+ Volume.State.Destroying
+ );
+
+ /** Serializes the structured event-description payloads to JSON. */
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ /**
+ * Timestamp of the last ASUP cycle start (or {@link Instant#EPOCH} if
none has run).
+ * The next run is due {@link OntapConfigurationManager#AsupIntervalHours}
after this
+ * instant. {@code volatile} so the scheduler thread's write is visible
without extra locking.
+ */
+ volatile Instant lastPushTime = Instant.EPOCH;
+
+ /** Single daemon thread that runs the ASUP pushes; owned by this manager.
*/
+ private ScheduledExecutorService asupScheduler;
+
+ /** The one pending run. Cancelled and re-booked whenever the schedule
changes. */
+ private ScheduledFuture<?> pendingRun;
+
+ @Inject
+ private PrimaryDataStoreDao storagePoolDao;
+ @Inject
+ private StoragePoolDetailsDao storagePoolDetailsDao;
+ @Inject
+ private VolumeDao volumeDao;
+ @Inject
+ private SnapshotDao snapshotDao;
+ @Inject
+ private VMSnapshotDao vmSnapshotDao;
+ @Inject
+ private ManagementService managementService;
+ @Inject
+ private ManagementServerHostDao managementServerHostDao;
+ @Inject
+ private MessageBus messageBus;
+
+ @Override
+ public boolean configure(String name, Map<String, Object> params) throws
ConfigurationException {
+ super.configure(name, params);
+ // React to edits of ontap.autosupport.interval so a new value
re-books the pending run
+ // instead of waiting for it to fire on the old schedule.
+ messageBus.subscribe(EventTypes.EVENT_CONFIGURATION_VALUE_EDIT,
this::onAsupConfigEdited);
+ return true;
+ }
+
+ @Override
+ public boolean start() {
+ asupScheduler = Executors.newSingleThreadScheduledExecutor(new
NamedThreadFactory("OntapAsup"));
+ logger.info("OntapAsupManager started; ASUP telemetry interval={}h",
+
getAsupIntervalHours(OntapConfigurationManager.AsupIntervalHours.value()));
+ scheduleNextRun();
+ return super.start();
+ }
+
+ @Override
+ public boolean stop() {
+ if (asupScheduler != null) {
+ asupScheduler.shutdownNow();
+ }
+ return super.stop();
+ }
+
+ /**
+ * Books the single pending run for the moment the configured interval
elapses, cancelling
+ * whatever was booked before. Called at start-up, after every run, and
whenever
+ * {@code ontap.autosupport.interval} is edited, so the live config always
decides the next run.
+ *
+ * <p>When the interval is {@link
OntapStorageConstants#ASUP_DISABLED_INTERVAL_HOURS}
+ * nothing is booked; setting a non-zero interval publishes a
configuration-edit event,
+ * which books a run again.</p>
+ */
+ private synchronized void scheduleNextRun() {
+ if (asupScheduler == null || asupScheduler.isShutdown()) {
+ return;
+ }
+ if (pendingRun != null) {
+ pendingRun.cancel(false);
+ pendingRun = null;
+ }
+ int intervalHours =
getAsupIntervalHours(OntapConfigurationManager.AsupIntervalHours.value());
+ if (intervalHours ==
OntapStorageConstants.ASUP_DISABLED_INTERVAL_HOURS) {
+ logger.debug("ONTAP ASUP: telemetry is disabled ({}={}); no run
scheduled.",
+ OntapStorageConstants.ASUP_INTERVAL_CONFIG_KEY,
+ OntapStorageConstants.ASUP_DISABLED_INTERVAL_HOURS);
+ return;
+ }
+ long delayMs = millisUntilNextPush();
+ pendingRun = asupScheduler.schedule(new OntapAsupTask(), delayMs,
TimeUnit.MILLISECONDS);
+ logger.debug("ONTAP ASUP: next telemetry push scheduled in {} ms.",
delayMs);
+ }
+
+ /**
+ * Milliseconds remaining until {@link #lastPushTime} plus the live
configured interval.
+ * Zero when a push is already overdue — for example after the interval is
shortened, the
+ * previous cycle ran longer than the interval, or on the very first run
when
+ * {@link #lastPushTime} is still {@link Instant#EPOCH}.
+ */
+ long millisUntilNextPush() {
+ Duration configuredInterval = Duration.ofHours(
+
getAsupIntervalHours(OntapConfigurationManager.AsupIntervalHours.value()));
+ Duration remaining =
configuredInterval.minus(Duration.between(lastPushTime, Instant.now()));
+ return remaining.isNegative() ? 0L : remaining.toMillis();
+ }
+
+ /**
+ * CloudStack publishes this after invalidating the config cache, so the
new value is already
+ * readable. Re-books the pending run: shortening the interval past the
due time schedules
+ * with zero delay, which pushes right away on the scheduler thread rather
than blocking the
+ * API thread that served {@code updateConfiguration}.
+ */
+ @SuppressWarnings("unchecked")
+ private void onAsupConfigEdited(String senderAddress, String subject,
Object args) {
+ if (!(args instanceof Ternary)) {
+ return;
+ }
+ String updatedKey = ((Ternary<String, ConfigKey.Scope, Long>)
args).first();
+ if
(!OntapConfigurationManager.AsupIntervalHours.key().equals(updatedKey)) {
+ return;
+ }
+ logger.debug("ONTAP ASUP: [{}] was updated; re-booking the next
push.", updatedKey);
+ scheduleNextRun();
+ }
+
+ /**
+ * One ASUP run inside a managed CloudStack context, which the DAO calls
made during the push
+ * require.
+ *
+ * <p>{@link #lastPushTime} is stamped at the start of the cycle so the
next wait is the
+ * remainder of the interval (work does not get added on top). The next
run is booked again
+ * in {@code finally} so the schedule never stalls.</p>
+ */
+ protected class OntapAsupTask extends ManagedContextRunnable {
+ @Override
+ protected void runInContext() {
+ try {
+ lastPushTime = Instant.now();
+ pushAsupTelemetry();
+ } catch (Exception e) {
+ logger.warn("ONTAP ASUP: unexpected error during periodic
push: {}", e.getMessage());
+ } finally {
+ scheduleNextRun();
+ }
+ }
+ }
+
+ /**
+ * Iterates all ONTAP-backed primary storage pools and pushes ASUP
telemetry for each.
+ *
+ * <p>Guarded by a {@link GlobalLock} so that, in a
multi-management-server deployment,
+ * only one node emits per cycle.</p>
+ */
+ protected void pushAsupTelemetry() {
+ List<StoragePoolVO> pools =
storagePoolDao.findPoolsByProvider(OntapStorageConstants.ONTAP_PLUGIN_NAME);
+ if (CollectionUtils.isEmpty(pools)) {
+ logger.debug("ONTAP ASUP: no ONTAP-backed storage pools found;
nothing to push.");
+ return;
+ }
+
+ GlobalLock lock =
GlobalLock.getInternLock(OntapStorageConstants.ASUP_GLOBAL_LOCK_NAME);
+ try {
+ if (!lock.lock(ASUP_LOCK_TIMEOUT_SECONDS)) {
+ logger.debug("ONTAP ASUP: another management server holds the
ASUP lock; skipping this cycle.");
+ return;
+ }
+ logger.debug("ONTAP ASUP: pushing telemetry for {} pool(s)
[CloudStack version={}]",
+ pools.size(), getCloudStackVersion());
+ Map<String, AsupClusterClient> clientsByStorageIp = new
HashMap<>();
+ for (StoragePoolVO pool : pools) {
+ pushAsupForStoragePool(pool, clientsByStorageIp);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Pushes the heartbeat (event-id 0) and pool (event-id 1) ASUP messages
for a single pool.
+ *
+ * <p>{@code clientsByStorageIp} is the per-cycle cache keyed by
management {@code storageIP}.
+ * A miss builds the strategy, GETs cluster info and node model, and sends
event-0. A hit
+ * reuses that client and skips those calls and event-0. Event-1 is always
sent. SVM name
+ * in the payload still comes from this pool's details.</p>
+ *
+ * <p>Best-effort: any failure is logged and swallowed.</p>
+ */
+ protected void pushAsupForStoragePool(StoragePoolVO pool, Map<String,
AsupClusterClient> clientsByStorageIp) {
+ try {
+ Map<String, String> details =
storagePoolDetailsDao.listDetailsKeyPairs(pool.getId());
+ if (details == null || details.isEmpty()) {
+ logger.warn("ONTAP ASUP: storage pool [{}] has no details;
skipping.", pool.getId());
+ return;
+ }
+
+ String storageIp = details.get(OntapStorageConstants.STORAGE_IP);
+ AsupClusterClient asupClusterClient =
StringUtils.isNotBlank(storageIp)
+ ? clientsByStorageIp.get(storageIp) : null;
+ boolean sendHeartbeat = asupClusterClient == null;
+ if (asupClusterClient == null) {
+ StorageStrategy strategy =
OntapStorageUtils.resolveStrategyFromPoolDetails(details);
+ Cluster cluster = strategy.getClusterInfo();
+ asupClusterClient = new AsupClusterClient(strategy, cluster);
+ if (StringUtils.isNotBlank(storageIp)) {
+ clientsByStorageIp.put(storageIp, asupClusterClient);
+ }
+ }
+
+ StorageStrategy strategy = asupClusterClient.strategy;
+ Cluster cluster = asupClusterClient.cluster;
+ String ontapVersion = strategy.getClusterVersion(cluster);
+ String clusterUuid = cluster != null ? cluster.getUuid() : null;
+ String cloudStackVersion = getCloudStackVersion();
+ String computerName = getComputerName();
+
+ if (sendHeartbeat) {
+ EmsApplicationLog heartbeat = buildBaseMessage(computerName,
cloudStackVersion);
+
heartbeat.setEventId(OntapStorageConstants.ASUP_EVENT_ID_HEARTBEAT);
+ heartbeat.setEventDescription(buildHeartbeatDescription(
+ cloudStackVersion, ontapVersion, cluster));
+ strategy.sendAsupMessage(heartbeat);
+ } else {
+ logger.debug("ONTAP ASUP: heartbeat already sent this cycle
for storage IP [{}]; skipping for pool [{}]",
+ storageIp, pool.getId());
+ }
+
+ // event-id 1: CloudStack storage pool -> backing ONTAP volume
mapping, once per pool.
+ // The description also includes disk usage and snapshot telemetry
+ EmsApplicationLog poolMessage = buildBaseMessage(computerName,
cloudStackVersion);
+
poolMessage.setEventId(OntapStorageConstants.ASUP_EVENT_ID_STORAGE_POOL);
+ poolMessage.setEventDescription(buildPoolDescription(pool,
details, clusterUuid));
+ strategy.sendAsupMessage(poolMessage);
+
+ logger.debug("ONTAP ASUP: pushed telemetry for pool [{}] (ONTAP
version={})",
+ pool.getId(), defaultUnknown(ontapVersion));
+ } catch (Exception e) {
+ // Best-effort telemetry; never propagate.
+ logger.warn("ONTAP ASUP: failed to push telemetry for pool [{}]:
{}", pool.getId(), e.getMessage());
+ }
+ }
+
+ /**
+ * Builds the heartbeat (event-id 0) description as a JSON object carrying
the CloudStack and
+ * ONTAP versions, cluster hardware model, the management-server operating
system platform,
+ * and the ONTAP cluster UUID.
+ * Example: {@code {"message":"CloudStack connected to ONTAP
cluster","cloudstackVersion":
+ * "4.23.0.0","platform":"Linux 5.15.0-91-generic
(amd64)","ontapVersion":"9.17.1",
+ *
"ontapClusterModel":"AFF-A400","ontapPlatformType":"performance","clusterUuid":"...",
+ * "managementServerCount":2}}
+ */
+ private String buildHeartbeatDescription(String cloudStackVersion, String
ontapVersion, Cluster cluster) {
+ Map<String, Object> payload = new LinkedHashMap<>();
+ payload.put(OntapStorageConstants.ASUP_MESSAGE,
OntapStorageConstants.ASUP_HEARTBEAT_MESSAGE);
+ payload.put(OntapStorageConstants.ASUP_CLOUDSTACK_VERSION,
defaultUnknown(cloudStackVersion));
+ payload.put(OntapStorageConstants.ASUP_PLATFORM, getOperatingSystem());
+ payload.put(OntapStorageConstants.ASUP_ONTAP_VERSION,
defaultUnknown(ontapVersion));
+ payload.put(OntapStorageConstants.ASUP_ONTAP_CLUSTER_MODEL,
+ defaultUnknown(cluster != null ? cluster.getModel() : null));
+ payload.put(OntapStorageConstants.ASUP_ONTAP_PLATFORM_TYPE,
+ defaultUnknown(cluster != null ? cluster.getPlatformType() :
null));
+ payload.put(OntapStorageConstants.ASUP_CLUSTER_UUID,
+ defaultUnknown(cluster != null ? cluster.getUuid() : null));
+ payload.put(OntapStorageConstants.ASUP_MANAGEMENT_SERVER_COUNT,
getManagementServerCount());
+ return toJson(payload);
+ }
+
+ /**
+ * Builds the pool description (event-id 1) as a JSON object combining the
backing-volume
+ * mapping, disk usage, and snapshot telemetry into a single EMS message.
+ *
+ * <p>Example: {@code {"message":"CloudStack storage pool backed by ONTAP
volume",
+ *
"poolName":"...","poolStatus":"Up","protocol":"nfs","clusterUuid":"...","svm":"...",
+ * "ontapVolumeUuid":"...","rootDiskCount":12,"dataDiskCount":18,
+ * "totalLogicalSizeBytes":322122547200,"multiPrimaryStoragePoolVm":false,
+ * "volumeSnapshotCount":5,"vmSnapshotCount":3}}</p>
+ */
+ private String buildPoolDescription(StoragePoolVO pool, Map<String,
String> details,
+ String clusterUuid) {
+ Map<String, Object> payload = new LinkedHashMap<>();
+ payload.put(OntapStorageConstants.ASUP_MESSAGE,
OntapStorageConstants.ASUP_POOL_MESSAGE);
+ payload.put(OntapStorageConstants.ASUP_POOL_NAME,
defaultUnknown(pool.getName()));
+ payload.put(OntapStorageConstants.ASUP_POOL_STATUS,
+ pool.getStatus() == null ? OntapStorageConstants.ASUP_UNKNOWN
: pool.getStatus().toString());
+ payload.put(OntapStorageConstants.ASUP_PROTOCOL,
defaultUnknown(details.get(OntapStorageConstants.PROTOCOL)));
+ payload.put(OntapStorageConstants.ASUP_CLUSTER_UUID,
defaultUnknown(clusterUuid));
+ payload.put(OntapStorageConstants.ASUP_SVM,
defaultUnknown(details.get(OntapStorageConstants.SVM_NAME)));
+ payload.put(OntapStorageConstants.ASUP_ONTAP_VOLUME_UUID,
defaultUnknown(details.get(OntapStorageConstants.VOLUME_UUID)));
+ List<VolumeVO> volumes = null;
+ try {
+ volumes = loadNonDestroyedVolumes(pool);
+ } catch (Exception e) {
+ logger.error("ONTAP ASUP: failed to load volumes for pool [{}]:
{}", pool.getId(), e.getMessage());
+ }
+ if (volumes != null) {
+ addStoragePoolUsage(pool, payload, volumes);
+ addSnapshotMetrics(pool, payload, volumes);
+ }
+ hasMultiPrimaryStoragePoolVm(pool, payload);
+ return toJson(payload);
+ }
+
+ /**
+ * Single CloudStack DB read of non-destroyed volumes on this pool. Shared
by usage and
+ * snapshot counts so the list is not queried three times. Never returns
null.
+ */
+ private List<VolumeVO> loadNonDestroyedVolumes(StoragePoolVO pool) {
+ List<VolumeVO> volumes =
volumeDao.findNonDestroyedVolumesByPoolId(pool.getId(), null);
+ return volumes == null ? java.util.Collections.emptyList() : volumes;
+ }
+
+ /**
+ * Computes pool usage from CloudStack's volume records and adds it to the
payload:
+ * <ul>
+ * <li>{@code rootDiskCount} - number of ROOT (boot) disks physically
on this pool</li>
+ * <li>{@code dataDiskCount} - number of DATADISK disks physically on
this pool</li>
+ * <li>{@code totalLogicalSizeBytes} - sum of those volumes'
provisioned (logical) sizes
+ * in bytes; for thin-provisioned volumes this is the logical size
requested at
+ * creation time, not the physical space consumed on ONTAP</li>
+ * </ul>
+ * {@code volumes} is the non-destroyed list for this pool (already
loaded). Only
+ * {@link #CS_VOLUME_STATES} are counted. Best-effort: any failure leaves
the usage
+ * fields out and never breaks telemetry.
+ */
+ private void addStoragePoolUsage(StoragePoolVO pool, Map<String, Object>
payload, List<VolumeVO> volumes) {
+ try {
+ // Only count volumes that definitely have a physical object on
the ONTAP FlexVolume.
+ // "Allocated" volumes have a pool_id row in the CS DB but ONTAP
provisioning has not
+ // yet been called, so including them would inflate counts and
provisioned size.
+ List<VolumeVO> cstackVolumes = volumes.stream()
+ .filter(v -> CS_VOLUME_STATES.contains(v.getState()))
+ .collect(java.util.stream.Collectors.toList());
+
+ long rootDiskCount = cstackVolumes.stream()
+ .filter(v ->
Volume.Type.ROOT.equals(v.getVolumeType())).count();
+ long dataDiskCount = cstackVolumes.stream()
+ .filter(v ->
Volume.Type.DATADISK.equals(v.getVolumeType())).count();
+
+ long totalLogicalSizeBytes = cstackVolumes.stream()
+ .mapToLong(v -> v.getSize() != null ? v.getSize() :
0L).sum();
Review Comment:
This allocates an intermediate list and then traverses it three more times.
Since this runs once per pool per interval, on very large pools this adds
avoidable CPU/GC overhead. Consider computing `rootDiskCount`, `dataDiskCount`,
and `totalLogicalSizeBytes` in a single pass over `volumes` (filtering by
`CS_VOLUME_STATES` inline) to reduce allocations and repeated streaming.
##########
engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java:
##########
@@ -96,6 +96,14 @@ public class VolumeDaoImpl extends GenericDaoBase<VolumeVO,
Long> implements Vol
protected static final String SELECT_HYPERTYPE_FROM_CLUSTER_VOLUME =
"SELECT c.hypervisor_type from volumes v, storage_pool s, cluster c where
v.pool_id = s.id and s.cluster_id = c.id and v.id = ?";
protected static final String SELECT_HYPERTYPE_FROM_ZONE_VOLUME = "SELECT
s.hypervisor from volumes v, storage_pool s where v.pool_id = s.id and v.id =
?";
protected static final String SELECT_POOLSCOPE = "SELECT s.scope from
storage_pool s, volumes v where s.id = v.pool_id and v.id = ?";
+ // Looks for a VM whose root disk is on this pool and whose data disk is
on another pool.
+ // LIMIT 1 stops after the first match so we do not load all volumes.
+ private static final String HAS_MULTI_PRIMARY_STORAGE_POOL_VM =
+ "SELECT 1 FROM volumes root INNER JOIN volumes data ON
data.instance_id = root.instance_id "
+ + "WHERE root.pool_id = ? AND root.volume_type = 'ROOT'
AND root.instance_id IS NOT NULL "
+ + "AND root.removed IS NULL AND root.state NOT IN
('Destroy', 'Expunged') "
+ + "AND data.volume_type = 'DATADISK' AND data.pool_id IS
NOT NULL AND data.pool_id <> ? "
+ + "AND data.removed IS NULL AND data.state NOT IN
('Destroy', 'Expunged') LIMIT 1";
Review Comment:
The DAO method is documented as “non-destroyed” volumes, but the SQL
currently includes volumes in transitional/pre-provisioning states (e.g.,
`Allocated`) because it only excludes `Destroy`/`Expunged`. That can
incorrectly flip `multiPrimaryStoragePoolVm=true` for VMs whose data disk
hasn’t actually been created/moved yet. Tighten the state predicates to align
with the intended “physically exists” semantics (e.g., include only the same
stable states you count elsewhere, or explicitly exclude
`Allocated`/upload-related states).
##########
plugins/storage/volume/ontap/src/main/java/org/apache/cloudstack/storage/asup/OntapAsupManager.java:
##########
@@ -0,0 +1,666 @@
+/*
+ * 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.asup;
+
+import com.cloud.cluster.ManagementServerHostVO;
+import com.cloud.cluster.dao.ManagementServerHostDao;
+import com.cloud.event.EventTypes;
+import com.cloud.server.ManagementService;
+import com.cloud.storage.Volume;
+import com.cloud.storage.SnapshotVO;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.SnapshotDao;
+import com.cloud.storage.dao.VolumeDao;
+import com.cloud.vm.snapshot.VMSnapshot;
+import com.cloud.vm.snapshot.VMSnapshotVO;
+import com.cloud.vm.snapshot.dao.VMSnapshotDao;
+import com.cloud.utils.Ternary;
+import com.cloud.utils.component.ManagerBase;
+import com.cloud.utils.concurrency.NamedThreadFactory;
+import com.cloud.utils.db.GlobalLock;
+import com.cloud.utils.net.NetUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.messagebus.MessageBus;
+import org.apache.cloudstack.managed.context.ManagedContextRunnable;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailsDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
+import org.apache.cloudstack.storage.feign.model.Cluster;
+import org.apache.cloudstack.storage.feign.model.EmsApplicationLog;
+import org.apache.cloudstack.storage.service.StorageStrategy;
+import org.apache.cloudstack.storage.utils.OntapConfigurationManager;
+import org.apache.cloudstack.storage.utils.OntapStorageConstants;
+import org.apache.cloudstack.storage.utils.OntapStorageUtils;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.inject.Inject;
+import javax.naming.ConfigurationException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Periodic ASUP (AutoSupport) telemetry pusher for the NetApp ONTAP plugin.
+ *
+ * <p>Each run pushes, for every ONTAP-backed primary storage pool, two
minimal EMS
+ * application-log messages to the backing ONTAP cluster:</p>
+ *
+ * <ul>
+ * <li><b>event-id 0 (heartbeat):</b> identifies the CloudStack deployment
(CloudStack
+ * version, management host) connected to the ONTAP cluster (ONTAP
cluster version).</li>
+ * <li><b>event-id 1 (pool):</b> maps the CloudStack storage pool to its
backing ONTAP
+ * volume - protocol (NFS/iSCSI), ONTAP FlexVolume UUID, SVM, disk
usage, and
+ * snapshot telemetry (counts by state, total provisioned size).</li>
+ * </ul>
+ *
+ * <p>Runs are booked one at a time: after each cycle starts, the next run is
scheduled for
+ * {@link OntapConfigurationManager#AsupIntervalHours} after that start, so
production
+ * intervals (hours) are start-to-start. REST work sits inside the interval;
it is not added
+ * after it. Editing the ONTAP ASUP interval re-books the pending run
immediately, with no
+ * management-server restart.</p>
+ */
+public class OntapAsupManager extends ManagerBase {
+ private static final int ASUP_LOCK_TIMEOUT_SECONDS = 5;
+
+ /**
+ * Volume states that guarantee a physical object exists on the ONTAP
FlexVolume.
+ * States like {@link Volume.State#Allocated} have a CloudStack DB row
pointing to this
+ * pool but ONTAP provisioning has not been called yet — they must be
excluded to avoid
+ * inflating disk counts and provisioned-size totals. Upload-family states
live on
+ * secondary storage, not on the primary ONTAP volume, so they are also
excluded.
+ */
+ private static final Set<Volume.State> CS_VOLUME_STATES = EnumSet.of(
+ Volume.State.Ready,
+ Volume.State.Snapshotting,
+ Volume.State.RevertSnapshotting,
+ Volume.State.Attaching,
+ Volume.State.Restoring,
+ Volume.State.Expunging,
+ Volume.State.Destroying
+ );
+
+ /** Serializes the structured event-description payloads to JSON. */
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ /**
+ * Timestamp of the last ASUP cycle start (or {@link Instant#EPOCH} if
none has run).
+ * The next run is due {@link OntapConfigurationManager#AsupIntervalHours}
after this
+ * instant. {@code volatile} so the scheduler thread's write is visible
without extra locking.
+ */
+ volatile Instant lastPushTime = Instant.EPOCH;
+
+ /** Single daemon thread that runs the ASUP pushes; owned by this manager.
*/
+ private ScheduledExecutorService asupScheduler;
+
+ /** The one pending run. Cancelled and re-booked whenever the schedule
changes. */
+ private ScheduledFuture<?> pendingRun;
+
+ @Inject
+ private PrimaryDataStoreDao storagePoolDao;
+ @Inject
+ private StoragePoolDetailsDao storagePoolDetailsDao;
+ @Inject
+ private VolumeDao volumeDao;
+ @Inject
+ private SnapshotDao snapshotDao;
+ @Inject
+ private VMSnapshotDao vmSnapshotDao;
+ @Inject
+ private ManagementService managementService;
+ @Inject
+ private ManagementServerHostDao managementServerHostDao;
+ @Inject
+ private MessageBus messageBus;
+
+ @Override
+ public boolean configure(String name, Map<String, Object> params) throws
ConfigurationException {
+ super.configure(name, params);
+ // React to edits of ontap.autosupport.interval so a new value
re-books the pending run
+ // instead of waiting for it to fire on the old schedule.
+ messageBus.subscribe(EventTypes.EVENT_CONFIGURATION_VALUE_EDIT,
this::onAsupConfigEdited);
+ return true;
+ }
+
+ @Override
+ public boolean start() {
+ asupScheduler = Executors.newSingleThreadScheduledExecutor(new
NamedThreadFactory("OntapAsup"));
+ logger.info("OntapAsupManager started; ASUP telemetry interval={}h",
+
getAsupIntervalHours(OntapConfigurationManager.AsupIntervalHours.value()));
+ scheduleNextRun();
+ return super.start();
+ }
+
+ @Override
+ public boolean stop() {
+ if (asupScheduler != null) {
+ asupScheduler.shutdownNow();
+ }
+ return super.stop();
+ }
+
+ /**
+ * Books the single pending run for the moment the configured interval
elapses, cancelling
+ * whatever was booked before. Called at start-up, after every run, and
whenever
+ * {@code ontap.autosupport.interval} is edited, so the live config always
decides the next run.
+ *
+ * <p>When the interval is {@link
OntapStorageConstants#ASUP_DISABLED_INTERVAL_HOURS}
+ * nothing is booked; setting a non-zero interval publishes a
configuration-edit event,
+ * which books a run again.</p>
+ */
+ private synchronized void scheduleNextRun() {
+ if (asupScheduler == null || asupScheduler.isShutdown()) {
+ return;
+ }
+ if (pendingRun != null) {
+ pendingRun.cancel(false);
+ pendingRun = null;
+ }
+ int intervalHours =
getAsupIntervalHours(OntapConfigurationManager.AsupIntervalHours.value());
+ if (intervalHours ==
OntapStorageConstants.ASUP_DISABLED_INTERVAL_HOURS) {
+ logger.debug("ONTAP ASUP: telemetry is disabled ({}={}); no run
scheduled.",
+ OntapStorageConstants.ASUP_INTERVAL_CONFIG_KEY,
+ OntapStorageConstants.ASUP_DISABLED_INTERVAL_HOURS);
+ return;
+ }
+ long delayMs = millisUntilNextPush();
+ pendingRun = asupScheduler.schedule(new OntapAsupTask(), delayMs,
TimeUnit.MILLISECONDS);
+ logger.debug("ONTAP ASUP: next telemetry push scheduled in {} ms.",
delayMs);
+ }
+
+ /**
+ * Milliseconds remaining until {@link #lastPushTime} plus the live
configured interval.
+ * Zero when a push is already overdue — for example after the interval is
shortened, the
+ * previous cycle ran longer than the interval, or on the very first run
when
+ * {@link #lastPushTime} is still {@link Instant#EPOCH}.
+ */
+ long millisUntilNextPush() {
+ Duration configuredInterval = Duration.ofHours(
+
getAsupIntervalHours(OntapConfigurationManager.AsupIntervalHours.value()));
+ Duration remaining =
configuredInterval.minus(Duration.between(lastPushTime, Instant.now()));
+ return remaining.isNegative() ? 0L : remaining.toMillis();
+ }
+
+ /**
+ * CloudStack publishes this after invalidating the config cache, so the
new value is already
+ * readable. Re-books the pending run: shortening the interval past the
due time schedules
+ * with zero delay, which pushes right away on the scheduler thread rather
than blocking the
+ * API thread that served {@code updateConfiguration}.
+ */
+ @SuppressWarnings("unchecked")
+ private void onAsupConfigEdited(String senderAddress, String subject,
Object args) {
+ if (!(args instanceof Ternary)) {
+ return;
+ }
+ String updatedKey = ((Ternary<String, ConfigKey.Scope, Long>)
args).first();
+ if
(!OntapConfigurationManager.AsupIntervalHours.key().equals(updatedKey)) {
+ return;
+ }
+ logger.debug("ONTAP ASUP: [{}] was updated; re-booking the next
push.", updatedKey);
+ scheduleNextRun();
+ }
+
+ /**
+ * One ASUP run inside a managed CloudStack context, which the DAO calls
made during the push
+ * require.
+ *
+ * <p>{@link #lastPushTime} is stamped at the start of the cycle so the
next wait is the
+ * remainder of the interval (work does not get added on top). The next
run is booked again
+ * in {@code finally} so the schedule never stalls.</p>
+ */
+ protected class OntapAsupTask extends ManagedContextRunnable {
+ @Override
+ protected void runInContext() {
+ try {
+ lastPushTime = Instant.now();
+ pushAsupTelemetry();
+ } catch (Exception e) {
+ logger.warn("ONTAP ASUP: unexpected error during periodic
push: {}", e.getMessage());
+ } finally {
+ scheduleNextRun();
+ }
+ }
+ }
+
+ /**
+ * Iterates all ONTAP-backed primary storage pools and pushes ASUP
telemetry for each.
+ *
+ * <p>Guarded by a {@link GlobalLock} so that, in a
multi-management-server deployment,
+ * only one node emits per cycle.</p>
+ */
+ protected void pushAsupTelemetry() {
+ List<StoragePoolVO> pools =
storagePoolDao.findPoolsByProvider(OntapStorageConstants.ONTAP_PLUGIN_NAME);
+ if (CollectionUtils.isEmpty(pools)) {
+ logger.debug("ONTAP ASUP: no ONTAP-backed storage pools found;
nothing to push.");
+ return;
+ }
+
+ GlobalLock lock =
GlobalLock.getInternLock(OntapStorageConstants.ASUP_GLOBAL_LOCK_NAME);
+ try {
+ if (!lock.lock(ASUP_LOCK_TIMEOUT_SECONDS)) {
+ logger.debug("ONTAP ASUP: another management server holds the
ASUP lock; skipping this cycle.");
+ return;
+ }
+ logger.debug("ONTAP ASUP: pushing telemetry for {} pool(s)
[CloudStack version={}]",
+ pools.size(), getCloudStackVersion());
+ Map<String, AsupClusterClient> clientsByStorageIp = new
HashMap<>();
+ for (StoragePoolVO pool : pools) {
+ pushAsupForStoragePool(pool, clientsByStorageIp);
+ }
+ } finally {
+ lock.unlock();
+ }
Review Comment:
`lock.unlock()` is executed even when `lock.lock(...)` returns false (early
`return`), which can attempt to unlock a lock this thread never acquired.
Additionally, `GlobalLock.getInternLock(...)` typically requires releasing the
reference (e.g., `releaseRef()`) to avoid leaking the interned lock object.
Track a `boolean locked` and only `unlock()` when true, and ensure the lock
reference is released in `finally` (even when not acquired).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
