CC-6863: Add created-at volume labels and warn users about old volumes

Author: gpanders

src/workerd/api/container.c++

@@ -4,12 +4,12 @@
 
 #include "container.h"
 
-#include <cmath>
-
 #include <workerd/api/http.h>
 #include <workerd/io/features.h>
 #include <workerd/io/io-context.h>
 
+#include <cmath>
+
 namespace workerd::api {
 
 // =======================================================================================
@@ -68,8 +68,7 @@ void Container::start(jsg::Lock& js, jsg::Optional<StartupOptions> maybeOptions)
         auto snapshot = list[i];
         double size = snapshots[i].size;
         JSG_REQUIRE(std::isfinite(size) && size >= 0 &&
-                size <= static_cast<double>((1ull << 53) - 1) &&
-                std::floor(size) == size,
+                size <= static_cast<double>((1ull << 53) - 1) && std::floor(size) == size,
             RangeError, "Snapshot size must be a non-negative integer <= Number.MAX_SAFE_INTEGER");
         snapshot.setId(snapshots[i].id);
         snapshot.setSize(static_cast<uint64_t>(size));
@@ -104,7 +103,8 @@ jsg::Promise<DirectorySnapshot> Container::snapshotDirectory(
 
   return IoContext::current()
       .awaitIo(js, req.send())
-      .then(js, [](jsg::Lock& js, capnp::Response<rpc::Container::SnapshotDirectoryResults> results) {
+      .then(
+          js, [](jsg::Lock& js, capnp::Response<rpc::Container::SnapshotDirectoryResults> results) {
     auto snapshot = results.getSnapshot();
     jsg::Optional<kj::String> name = kj::none;
     auto snapshotName = snapshot.getName();

src/workerd/api/container.h

@@ -87,7 +87,8 @@ class Container: public jsg::Object {
   jsg::Promise<void> interceptOutboundHttp(
       jsg::Lock& js, kj::String addr, jsg::Ref<Fetcher> binding);
   jsg::Promise<void> interceptAllOutboundHttp(jsg::Lock& js, jsg::Ref<Fetcher> binding);
-  jsg::Promise<DirectorySnapshot> snapshotDirectory(jsg::Lock& js, SnapshotDirectoryOptions options);
+  jsg::Promise<DirectorySnapshot> snapshotDirectory(
+      jsg::Lock& js, SnapshotDirectoryOptions options);
 
   // TODO(containers): listenTcp()
 

src/workerd/server/container-client.c++

@@ -25,6 +25,8 @@
 #include <kj/exception.h>
 #include <kj/string.h>
 
+#include <atomic>
+
 namespace workerd::server {
 
 namespace {
@@ -35,13 +37,17 @@ constexpr uint16_t SIDECAR_INGRESS_PORT = 39001;
 constexpr uint64_t MAX_JSON_RESPONSE_SIZE = 16ULL * 1024 * 1024;
 
 constexpr kj::StringPtr SNAPSHOT_VOLUME_PREFIX = "workerd-snap-"_kj;
+constexpr kj::StringPtr SNAPSHOT_VOLUME_CREATED_AT_LABEL = "dev.workerd.snapshot-created-at"_kj;
+constexpr auto SNAPSHOT_STALE_AGE = 30 * kj::DAYS;
 
 // Maximum size of a snapshot tar archive held in memory during snapshot create/restore.
 constexpr size_t MAX_SNAPSHOT_TAR_SIZE = 1ULL * 1024 * 1024 * 1024;  // 1 GiB
 
 static_assert(static_cast<double>(MAX_SNAPSHOT_TAR_SIZE) == MAX_SNAPSHOT_TAR_SIZE,
     "MAX_SNAPSHOT_TAR_SIZE must be exactly representable as double");
 
+std::atomic<bool> staleSnapshotVolumeCheckScheduled = false;
+
 // Validate a snapshot directory path. Rejects relative paths, embedded null bytes,
 // and path traversal components (".."). Returns the validated parent directory path
 // (with leading '/') suitable for the Docker archive API.
@@ -357,6 +363,64 @@ kj::Promise<DockerBinaryResponse> dockerApiBinaryRequest(kj::Network& network,
       bodyBytes, "application/x-tar"_kj, maxResponseSize);
 }
 
+kj::String currentSnapshotVolumeTimestamp() {
+  return kj::str((kj::systemPreciseCalendarClock().now() - kj::UNIX_EPOCH) / kj::SECONDS);
+}
+
+kj::Maybe<int64_t> tryGetSnapshotCreatedAt(capnp::JsonValue::Reader labels) {
+  if (!labels.isObject()) {
+    return kj::none;
+  }
+
+  for (auto field: labels.getObject()) {
+    if (field.getName() != SNAPSHOT_VOLUME_CREATED_AT_LABEL) {
+      continue;
+    }
+
+    auto value = field.getValue();
+    if (!value.isString()) {
+      return kj::none;
+    }
+    return value.getString().tryParseAs<int64_t>();
+  }
+
+  return kj::none;
+}
+
+kj::Promise<void> warnAboutStaleSnapshotVolumes(kj::Network& network, kj::String dockerPath) {
+  capnp::JsonCodec codec;
+  codec.handleByAnnotation<docker_api::Docker::VolumeListFilters>();
+  capnp::MallocMessageBuilder filterMessage;
+  auto filters = filterMessage.initRoot<docker_api::Docker::VolumeListFilters>();
+  auto names = filters.initName(1);
+  names.set(0, SNAPSHOT_VOLUME_PREFIX);
+
+  auto response = co_await dockerApiRequest(network, kj::mv(dockerPath), kj::HttpMethod::GET,
+      kj::str("/volumes?filters=", kj::encodeUriComponent(codec.encode(filters))));
+  if (response.statusCode != 200) {
+    co_return;
+  }
+
+  auto message = decodeJsonResponse<docker_api::Docker::VolumeListResponse>(response.body);
+  auto root = message->getRoot<docker_api::Docker::VolumeListResponse>();
+  auto now = kj::systemPreciseCalendarClock().now();
+  kj::Vector<kj::String> staleVolumes;
+
+  for (auto volume: root.getVolumes()) {
+    KJ_IF_SOME(createdAtSeconds, tryGetSnapshotCreatedAt(volume.getLabels())) {
+      auto createdAt = kj::UNIX_EPOCH + createdAtSeconds * kj::SECONDS;
+      if (now - createdAt >= SNAPSHOT_STALE_AGE) {
+        staleVolumes.add(kj::str(volume.getName()));
+      }
+    }
+  }
+
+  if (staleVolumes.size() > 0) {
+    KJ_LOG(WARNING, "the following snapshot volumes were created 30+ days ago and may be stale",
+        kj::strArray(staleVolumes, ", "));
+  }
+}
+
 }  // namespace
 
 ContainerClient::ContainerClient(capnp::ByteStreamFactory& byteStreamFactory,
@@ -381,7 +445,14 @@ ContainerClient::ContainerClient(capnp::ByteStreamFactory& byteStreamFactory,
       waitUntilTasks(waitUntilTasks),
       pendingCleanup(kj::mv(pendingCleanup).fork()),
       cleanupCallback(kj::mv(cleanupCallback)),
-      channelTokenHandler(channelTokenHandler) {}
+      channelTokenHandler(channelTokenHandler) {
+  if (!staleSnapshotVolumeCheckScheduled.exchange(true, std::memory_order_relaxed)) {
+    waitUntilTasks.add(warnAboutStaleSnapshotVolumes(network, kj::str(this->dockerPath))
+                           .catch_([](kj::Exception&& e) {
+      KJ_LOG(WARNING, "failed to inspect snapshot volumes for staleness", e);
+    }));
+  }
+}
 
 ContainerClient::~ContainerClient() noexcept(false) {
   stopEgressListener();
@@ -1055,6 +1126,9 @@ kj::Promise<void> ContainerClient::createDockerVolume(kj::StringPtr volumeName)
   capnp::MallocMessageBuilder message;
   auto req = message.initRoot<docker_api::Docker::VolumeCreateRequest>();
   req.setName(volumeName);
+  auto labels = req.initLabels().initObject(1);
+  labels[0].setName(SNAPSHOT_VOLUME_CREATED_AT_LABEL);
+  labels[0].initValue().setString(currentSnapshotVolumeTimestamp());
 
   auto response = co_await dockerApiRequest(network, kj::str(dockerPath), kj::HttpMethod::POST,
       kj::str("/volumes/create"), codec.encode(req));
@@ -1334,6 +1408,7 @@ kj::Promise<void> ContainerClient::snapshotDirectory(SnapshotDirectoryContext co
       "': ", putResponse.statusCode);
 
   volumeCommitted = true;
+  KJ_LOG(INFO, "created snapshot volume", volumeName, dir, tarSize);
 
   // Populate the capnp response.
   auto result = context.getResults().initSnapshot();

src/workerd/server/docker-api.capnp

@@ -317,6 +317,12 @@ struct Docker {
   # Volume create request (POST /volumes/create)
   struct VolumeCreateRequest {
     name @0 :Text $Json.name("Name");
+    labels @1 :Json.Value $Json.name("Labels");
+  }
+
+  # Volume list filters query parameter (GET /volumes?filters=...)
+  struct VolumeListFilters {
+    name @0 :List(Text) $Json.name("name");
   }
 
   # Volume list response (GET /volumes)
@@ -326,6 +332,7 @@ struct Docker {
 
     struct Volume {
       name @0 :Text $Json.name("Name");
+      labels @1 :Json.Value $Json.name("Labels");
     }
   }
 }

src/workerd/server/tests/container-client/test.js

@@ -492,6 +492,10 @@ export class DurableObjectExample extends DurableObject {
   }
 
   async restoreTransferredSnapshot(snapshot) {
+    assert.ok(snapshot.id, 'snapshot must have a non-empty id');
+    assert.ok(snapshot.size > 0, 'snapshot must have a positive size');
+    assert.strictEqual(snapshot.dir, '/app/data');
+
     const container = this.ctx.container;
     if (container.running) {
       const monitor = container.monitor().catch((_err) => {});