feat(usecases): add cns:// source retrieval and DataNexus workflow recipes

Justfile

@@ -12,7 +12,9 @@ gcp_region := env_var_or_default("LOGSTORY_GCP_REGION", if chronicle_region == "
 api_type := env_var_or_default("LOGSTORY_API_TYPE", "rest")
 forwarder_name := env_var_or_default("LOGSTORY_FORWARDER_NAME", "Logstory-REST-Forwarder")
 secret_name := env_var_or_default("LOGSTORY_SECRET_NAME", "chronicle-api-key")
-usecases_bucket := env_var_or_default("LOGSTORY_USECASES_BUCKET", "gs://logstory-usecases-20241216")
+usecases_bucket := env_var_or_default("LOGSTORY_USECASES_BUCKET", "cns:///cns/is-d/home/dandye/logstory_usecases_20260918")
+cns_dir := env_var_or_default("LOGSTORY_CNS_DIR", "/cns/is-d/home/dandye/logstory_usecases_20260918")
+datanexus_entry := env_var_or_default("LOGSTORY_DATANEXUS_ENTRY", "mldataset.cloudsec_adoption.logstory_usecases_2026_09_18")
 timestamp_delta := env_var_or_default("LOGSTORY_TIMESTAMP_DELTA", "1d")
 venv_dir := "venv"
 
@@ -98,7 +100,12 @@ default:
             ('pre-commit-run', '', 'Run pre-commit hooks on all files'),
             ('pre-commit-update', '', 'Update pre-commit hooks to latest versions'),
         ]),
-        ('Usecases & GCS Storage', C.BCYAN, C.CYAN, [
+        ('Usecases, CNS & GCS Storage', C.BCYAN, C.CYAN, [
+            ('usecase-publish-cns', '<usecase> [cns_dir]', 'Sync a single local usecase to Google Colossus (CNS)'),
+            ('usecase-publish-all-cns', '[dir] [cns_dir]', 'Sync all local usecases to Google Colossus (CNS)'),
+            ('usecase-list-cns', '[cns_dir]', 'List usecases stored in Google Colossus (CNS)'),
+            ('datanexus-register-dryrun', '[entry] [variant] [cns_dir]', 'Dry-run DataNexus / ML Catalog variant registration via mlc'),
+            ('datanexus-register', '[entry] [variant] [cns_dir]', 'Register Colossus directory variant in DataNexus via mlc'),
             ('usecase-publish', '<usecase> [bucket]', 'Sync a single local usecase to Google Cloud Storage bucket'),
             ('usecase-publish-all', '[dir] [bucket]', 'Sync all local usecases to Google Cloud Storage bucket'),
             ('usecase-list-gcs', '[bucket]', 'List usecases stored in Google Cloud Storage bucket'),
@@ -757,9 +764,58 @@ cloudrun-help:
     @echo "  just cloudrun-delete-all"
 
 # ==============================================================================
-# Usecases and Storage Bucket Management
+# Usecases, CNS, DataNexus, and Storage Bucket Management
 # ==============================================================================
 
+# Publish a single local usecase to Google Colossus (CNS)
+usecase-publish-cns usecase target_cns=cns_dir:
+    @if [ -d "usecases/{{ usecase }}" ]; then \
+        echo "Publishing usecase '{{ usecase }}' to {{ target_cns }}/{{ usecase }}..."; \
+        fileutil mkdir -p "{{ target_cns }}/{{ usecase }}"; \
+        fileutil cp -f -R "usecases/{{ usecase }}/*" "{{ target_cns }}/{{ usecase }}/"; \
+    elif [ -d "src/logstory/usecases/{{ usecase }}" ]; then \
+        echo "Publishing usecase '{{ usecase }}' to {{ target_cns }}/{{ usecase }}..."; \
+        fileutil mkdir -p "{{ target_cns }}/{{ usecase }}"; \
+        fileutil cp -f -R "src/logstory/usecases/{{ usecase }}/*" "{{ target_cns }}/{{ usecase }}/"; \
+    else \
+        echo "Error: Usecase '{{ usecase }}' not found in usecases/ or src/logstory/usecases/"; \
+        exit 1; \
+    fi
+    @echo "Usecase '{{ usecase }}' published to CNS successfully!"
+
+# Publish all local usecases to Google Colossus (CNS)
+usecase-publish-all-cns local_dir="usecases" target_cns=cns_dir:
+    @if [ ! -d "{{ local_dir }}" ]; then \
+        echo "Error: Local directory '{{ local_dir }}' does not exist."; \
+        exit 1; \
+    fi
+    @echo "Publishing all usecases from '{{ local_dir }}' to {{ target_cns }}..."
+    fileutil mkdir -p "{{ target_cns }}"
+    fileutil cp -f -R "{{ local_dir }}/*" "{{ target_cns }}/"
+    @echo "All usecases published to CNS successfully!"
+
+# List usecases stored in Google Colossus (CNS)
+usecase-list-cns target_cns=cns_dir:
+    fileutil ls "{{ target_cns }}"
+
+# Dry-run registering the Colossus directory as a variant in DataNexus / ML Catalog
+datanexus-register-dryrun entry=datanexus_entry variant="20260918" target_cns=cns_dir format="AUTO":
+    SOURCE_PROTO="$$(/google/bin/releases/ml-catalog/tools/mlc generate file-source "{{ target_cns }}/*" "{{ format }}")" && \
+    /google/bin/releases/ml-catalog/tools/mlc create \
+        --dry_run \
+        "{{ entry }}" "{{ variant }}" \
+        --aliases_assign="latest" \
+        --source="$$SOURCE_PROTO"
+
+# Commit and set default variant in DataNexus / ML Catalog
+datanexus-register entry=datanexus_entry variant="20260918" target_cns=cns_dir format="AUTO":
+    SOURCE_PROTO="$$(/google/bin/releases/ml-catalog/tools/mlc generate file-source "{{ target_cns }}/*" "{{ format }}")" && \
+    /google/bin/releases/ml-catalog/tools/mlc create \
+        "{{ entry }}" "{{ variant }}" \
+        --set_default \
+        --aliases_assign="latest" \
+        --source="$$SOURCE_PROTO"
+
 # Publish a single local usecase to the Google Cloud Storage bucket
 usecase-publish usecase bucket=usecases_bucket:
     @if [ -d "usecases/{{ usecase }}" ]; then \

src/logstory/logstory.py

@@ -39,7 +39,7 @@
 
 UTC = datetime.UTC
 
-DEFAULT_BUCKET = "gs://logstory-usecases-20241216"
+DEFAULT_BUCKET = "cns:///cns/is-d/home/dandye/logstory_usecases_20260918"
 
 
 def version_callback(value: bool):
@@ -161,19 +161,34 @@ def get_auto_get_default():
   return auto_get_value in ("true", "1", "yes", "on")
 
 
+def _normalize_cns_path(raw_path: str) -> str:
+  """Normalize a CNS URI payload or path into a canonical /cns/... directory path."""
+  cleaned = raw_path.strip().rstrip("*").rstrip("/")
+  if cleaned.startswith("/cns/"):
+    return cleaned
+  if cleaned.startswith("cns/"):
+    return f"/{cleaned}"
+  cleaned = cleaned.lstrip("/")
+  return f"/cns/{cleaned}"
+
+
 def parse_usecase_source(source_uri: str) -> tuple[str, str]:
   """Parse a usecase source URI and return (source_type, identifier).
 
   Args:
-    source_uri: URI like 'gs://bucket-name' or bare bucket name
+    source_uri: URI like 'cns:///cns/...', 'gs://bucket-name', or bare bucket name
 
   Returns:
     Tuple of (source_type, identifier) where:
-    - source_type: 'gcs', 'git', etc.
-    - identifier: bucket name, repo URL, etc.
+    - source_type: 'cns', 'gcs', 'git', etc.
+    - identifier: CNS path, bucket name, repo URL, etc.
   """
   source_uri = source_uri.strip()
 
+  if source_uri.startswith("cns://"):
+    return ("cns", _normalize_cns_path(source_uri[6:]))
+  if source_uri.startswith("/cns/"):
+    return ("cns", _normalize_cns_path(source_uri))
   if source_uri.startswith("gs://"):
     return ("gcs", source_uri[5:])  # Remove 'gs://' prefix
   if source_uri.startswith("git@") or source_uri.endswith(".git"):
@@ -258,7 +273,10 @@ def get_usecases_buckets():
 UsecasesBucketOption = typer.Option(
     None,
     "--usecases-bucket",
-    help="Usecase source URI (gs://bucket, git@repo, etc.) - overrides config list",
+    help=(
+        "Usecase source URI (cns:///cns/..., gs://bucket, file:///dir, etc.) - "
+        "overrides config list"
+    ),
 )
 
 LocalFileOutputOption = typer.Option(
@@ -413,6 +431,8 @@ def _get_blobs(source_uri, usecase=None):
   """Get blobs from usecase source, supporting multiple source types."""
   source_type, identifier = parse_usecase_source(source_uri)
 
+  if source_type == "cns":
+    return _get_cns_blobs(identifier, usecase)
   if source_type == "gcs":
     return _get_gcs_blobs(identifier, usecase)
   if source_type == "git":
@@ -454,6 +474,119 @@ def _get_gcs_blobs(bucket_name, usecase=None):
   return blobs
 
 
+class _CnsBlob:
+  """Blob object for Google Colossus (CNS) operations via fileutil."""
+
+  def __init__(self, name: str, cns_path: str):
+    self.name = name
+    self._cns_path = cns_path
+
+  def download_to_filename(self, destination: str):
+    """Copy file from CNS to local destination using fileutil cp -f."""
+    cmd = ["fileutil", "cp", "-f", self._cns_path, destination]
+    gfs_user = os.getenv("LOGSTORY_GFS_USER")
+    if gfs_user:
+      cmd.insert(1, f"--gfs_user={gfs_user}")
+    try:
+      subprocess.run(  # nosec B603 B607 # noqa: S603
+          cmd,
+          check=True,
+          capture_output=True,
+          text=True,
+      )
+    except FileNotFoundError as e:
+      raise RuntimeError(
+          "fileutil binary not found in PATH; required for cns:// sources"
+      ) from e
+    except subprocess.CalledProcessError as e:
+      stderr = (e.stderr or "").strip()
+      raise RuntimeError(
+          f"Failed to copy CNS file '{self._cns_path}' to '{destination}': {stderr}"
+      ) from e
+
+
+def _get_cns_blobs(directory_path: str, usecase: str | None = None):
+  """Get blobs from Google Colossus (CNS) via fileutil, mimicking GCS blob interface."""
+  directory_path = _normalize_cns_path(directory_path)
+  gfs_user = os.getenv("LOGSTORY_GFS_USER")
+  base_cmd = ["fileutil"]
+  if gfs_user:
+    base_cmd.append(f"--gfs_user={gfs_user}")
+
+  if usecase:
+    usecase_path = f"{directory_path}/{usecase}"
+    cmd = [*base_cmd, "ls", "-l", "-R", usecase_path]
+    try:
+      result = subprocess.run(  # nosec B603 B607 # noqa: S603
+          cmd,
+          check=True,
+          capture_output=True,
+          text=True,
+      )
+    except FileNotFoundError as e:
+      raise RuntimeError(
+          "fileutil binary not found in PATH; required for cns:// sources"
+      ) from e
+    except subprocess.CalledProcessError as e:
+      stderr = (e.stderr or "").strip()
+      raise RuntimeError(
+          f"Usecase directory does not exist or is inaccessible on CNS: {usecase_path}"
+          f" ({stderr})"
+      ) from e
+
+    blobs = []
+    prefix_to_strip = f"{directory_path}/"
+    for line in result.stdout.splitlines():
+      line = line.strip()
+      if not line or not line.startswith("-"):
+        continue
+      parts = line.split(None, 7)
+      if len(parts) < 8:
+        continue
+      cns_file_path = parts[7]
+      if cns_file_path.startswith(prefix_to_strip):
+        blob_name = cns_file_path[len(prefix_to_strip) :]
+      else:
+        blob_name = os.path.relpath(cns_file_path, directory_path).replace(os.sep, "/")
+      blobs.append(_CnsBlob(blob_name, cns_file_path))
+    return blobs
+
+  # Return top-level directories (used for listing available usecases)
+  cmd = [*base_cmd, "ls", "-l", directory_path]
+  try:
+    result = subprocess.run(  # nosec B603 B607 # noqa: S603
+        cmd,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+  except FileNotFoundError as e:
+    raise RuntimeError(
+        "fileutil binary not found in PATH; required for cns:// sources"
+    ) from e
+  except subprocess.CalledProcessError as e:
+    stderr = (e.stderr or "").strip()
+    raise RuntimeError(
+        f"CNS directory does not exist or is inaccessible: {directory_path} ({stderr})"
+    ) from e
+
+  prefixes = []
+  for line in result.stdout.splitlines():
+    line = line.strip()
+    if not line or not line.startswith("d"):
+      continue
+    parts = line.split(None, 7)
+    if len(parts) < 8:
+      continue
+    full_path = parts[7].rstrip("/")
+    if full_path == directory_path:
+      continue
+    item_name = full_path.rsplit("/", 1)[-1]
+    prefixes.append(f"{item_name}/")
+
+  return _FileBlobCollection([_FileBlobPage(prefixes)])
+
+
 class _FileBlob:
   """Mock blob object for file system operations."""
 

tests/test_cli.py

@@ -26,11 +26,14 @@
 from typer.testing import CliRunner
 
 from logstory.logstory import (
+    DEFAULT_BUCKET,
+    _CnsBlob,
     _download_all_usecases,
     _download_usecase,
     _FileBlob,
     _get_all_source_directories,
     _get_blobs,
+    _get_cns_blobs,
     _get_file_blobs,
     _get_gcs_blobs,
     _get_logtypes,
@@ -171,10 +174,27 @@ def test_load_env_file(self):
 
 
 class TestSourceParsingAndBlobMocking:
-  """Test parsing GCS/file sources and blob collections."""
+  """Test parsing CNS/GCS/file sources and blob collections."""
+
+  def test_default_bucket_is_cns(self):
+    """Test DEFAULT_BUCKET defaults to the Colossus CNS directory."""
+    assert DEFAULT_BUCKET == "cns:///cns/is-d/home/dandye/logstory_usecases_20260918"
+    with patch.dict(os.environ, {}, clear=True):
+      assert get_usecases_buckets() == [
+          "cns:///cns/is-d/home/dandye/logstory_usecases_20260918"
+      ]
 
   def test_parse_usecase_source(self):
     """Test source URI parsing."""
+    assert parse_usecase_source(
+        "cns:///cns/is-d/home/dandye/logstory_usecases_20260918"
+    ) == ("cns", "/cns/is-d/home/dandye/logstory_usecases_20260918")
+    assert parse_usecase_source(
+        "cns://is-d/home/dandye/logstory_usecases_20260918"
+    ) == ("cns", "/cns/is-d/home/dandye/logstory_usecases_20260918")
+    assert parse_usecase_source(
+        "/cns/is-d/home/dandye/logstory_usecases_20260918/*"
+    ) == ("cns", "/cns/is-d/home/dandye/logstory_usecases_20260918")
     assert parse_usecase_source("gs://my-bucket/usecases") == (
         "gcs",
         "my-bucket/usecases",
@@ -210,6 +230,62 @@ def test_file_blob_and_collection(self):
       fb.download_to_filename(str(dest))
       assert dest.read_text(encoding="utf-8") == "sample content"
 
+  @patch("subprocess.run")
+  def test_cns_blob_and_collection(self, mock_run):
+    """Test _CnsBlob and _get_cns_blobs directory and recursive file listing."""
+    # 1. Directory listing (usecase=None)
+    mock_run.return_value = MagicMock(
+        stdout=(
+            "drwxrwx--- 1 dandye empty 0 2026/09/18 23:34:56"
+            " /cns/is-d/home/dandye/logstory_usecases_20260918/NETWORK_ANALYSIS\n"
+            "-rw-rw---- 1 dandye empty 3664 2026/09/18 23:34:57"
+            " /cns/is-d/home/dandye/logstory_usecases_20260918/README_template.md\n"
+            "drwxrwx--- 1 dandye empty 0 2026/09/18 23:35:07"
+            " /cns/is-d/home/dandye/logstory_usecases_20260918/RULES_SEARCH_WORKSHOP\n"
+        )
+    )
+    collection = _get_cns_blobs("/cns/is-d/home/dandye/logstory_usecases_20260918")
+    assert len(collection.pages) == 1
+    assert collection.pages[0].prefixes == [
+        "NETWORK_ANALYSIS/",
+        "RULES_SEARCH_WORKSHOP/",
+    ]
+
+    # 2. Recursive file listing for specific usecase
+    mock_run.return_value = MagicMock(
+        stdout=(
+            "drwxrwx--- 1 dandye empty 0 2026/09/18 23:34:56"
+            " /cns/is-d/home/dandye/logstory_usecases_20260918/NETWORK_ANALYSIS\n"
+            "-rw-rw---- 1 dandye empty 2540751 2026/09/18 23:34:54"
+            " /cns/is-d/home/dandye/logstory_usecases_20260918/NETWORK_ANALYSIS/EVENTS/BRO_JSON.log\n"
+        )
+    )
+    blobs = _get_cns_blobs(
+        "/cns/is-d/home/dandye/logstory_usecases_20260918",
+        usecase="NETWORK_ANALYSIS",
+    )
+    assert len(blobs) == 1
+    assert isinstance(blobs[0], _CnsBlob)
+    assert blobs[0].name == "NETWORK_ANALYSIS/EVENTS/BRO_JSON.log"
+
+    # 3. Download via _CnsBlob with LOGSTORY_GFS_USER
+    mock_run.reset_mock()
+    with patch.dict(os.environ, {"LOGSTORY_GFS_USER": "my-group"}):
+      blobs[0].download_to_filename("/tmp/BRO_JSON.log")
+    mock_run.assert_called_once_with(
+        [
+            "fileutil",
+            "--gfs_user=my-group",
+            "cp",
+            "-f",
+            "/cns/is-d/home/dandye/logstory_usecases_20260918/NETWORK_ANALYSIS/EVENTS/BRO_JSON.log",
+            "/tmp/BRO_JSON.log",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
   @patch("logstory.logstory.storage.Client")
   def test_get_gcs_blobs(self, mock_storage_client):
     """Test _get_gcs_blobs with GCS client."""
@@ -225,7 +301,12 @@ def test_get_gcs_blobs(self, mock_storage_client):
     mock_bucket.list_blobs.assert_called_once_with(prefix="UC1")
 
   def test_get_blobs_dispatcher(self):
-    """Test _get_blobs dispatching to GCS or file."""
+    """Test _get_blobs dispatching to CNS, GCS, or file."""
+    with patch("logstory.logstory._get_cns_blobs", return_value=["cns_blob"]):
+      assert _get_blobs("cns:///cns/is-d/home/dandye/logstory_usecases_20260918") == [
+          "cns_blob"
+      ]
+
     with patch("logstory.logstory._get_gcs_blobs", return_value=["gcs_blob"]):
       assert _get_blobs("gs://bucket1") == ["gcs_blob"]