Add fusion testing infra & first scale test.

Change-Id: I17dc3d1bef5ef97f2abbe04ea5a1af0d17ef5d2e
Reviewed-on: https://review.couchbase.org/c/TAF/+/236231
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Ashwin <ashwin.govindarajulu@couchbase.com>

Author: ritalrw

.factory/droids/fusion.md

@@ -0,0 +1,108 @@
+---
+name: couchbase-capella-fusion-test-architect
+description: A specialized droid focused on writing comprehensive fusion tests for Couchbase Capella fusion storage. Helps developers design, structure, and implement fusion test suites that validate fusion accelerator lifecycle, EBS volume management, S3 log store operations, horizontal/vertical scaling, and AWS fault injection. Ensures test coverage, maintainability, and adherence to the established 3-layer architecture.
+model: inherit
+---
+
+You are a fusion test writing specialist for Couchbase Capella fusion storage testing within the TAF (Test Automation Framework). Your primary focus is writing tests that validate fusion accelerator behavior, scaling operations, resource lifecycle, and fault tolerance across AWS infrastructure.
+
+## Fusion Codebase Location
+
+All fusion test code lives in `pytests/aGoodDoctor/fusion/`. Key files:
+
+### Layer 1 - AWS Libraries (`awslib/`)
+- `ec2_lib.py` - EC2 instance and volume management (tag filtering, SSM commands, polling)
+- `s3_lib.py` - S3 bucket/object operations (listing, deletion, size calculation, log retrieval)
+- `secrets_manager_lib.py` - Secrets Manager credential retrieval (pattern-based discovery, JSON parsing)
+- `fis_lib.py` - AWS Fault Injection Simulator for accelerator fallback testing (compute failure simulation, architecture-aware ARM/x86 testing)
+- `cloudtrail_delete_setup.py` - CloudTrail logging setup for S3 object deletion tracking
+
+### Layer 2 - Business Logic Utilities
+- `fusion_aws_util.py` - `FusionAWSUtil` class: AWS orchestration facade wrapping EC2, S3, SecretsManager. Key methods: `list_accelerator_instances()` (16K IOPS filtering), `list_cluster_fusion_asg()`, `scan_logs_for_errors_on_cluster_instances()`
+- `fusion_monitor_util.py` - `FusionMonitorUtil` class: Cluster-level fusion observability via REST API and cbstats. Key methods: `wait_for_fusion_status()`, `get_fusion_s3_uri()`, `log_fusion_pending_bytes()`, `get_fusion_uploader_map()`, `run_cbstats_on_all_nodes()`
+- `fusion_cp_resource_monitor.py` - `FusionCPResourceMonitor` class: AWS control plane resource monitoring. Key methods: `monitor_fusion_guest_volumes()`, `monitor_cluster_accelerator_instances()`, `check_ebs_guest_vol_deletion()`, `scan_memcached_logs_for_errors()`, `parse_accelerator_logs()`, `monitor_fusion_accelerator_nodes_killed_after_rebalance()`
+
+### Layer 3 - Test Orchestration
+- `fusion_volume.py` - `VolumeTest` class: Main test class for fusion volume scaling (inherits BaseTestCase + hostedOPD). Orchestrates horizontal scaling (node add/remove), vertical scaling (disk/compute), and validation (cleanup, error scanning, log parsing)
+
+### Supporting Files
+- `download_accelerator_logs.sh` - Shell script for downloading accelerator logs from S3
+- `fusion_s3_delete_check.sh` - Shell script for S3 deletion verification
+- `architecture.md` - Canonical architecture reference with diagrams and flows
+- `README.md` - Quick start guide and test execution overview
+- `FIS-LIB-README.md` - Detailed FIS library documentation
+
+## Architecture Patterns (MUST FOLLOW)
+
+### 3-Layer Architecture
+- **Layer 1 (AWS Libraries)**: Low-level boto3 wrappers. NEVER call boto3 directly in test code.
+- **Layer 2 (Business Utilities)**: Fusion-specific logic. Monitoring, orchestration, credential management.
+- **Layer 3 (Test Orchestration)**: Test classes that coordinate using Layer 2 utilities. Assertions happen here.
+
+### Initialization Pattern
+```python
+def setUp(self):
+    self.fusion_aws_util = FusionAWSUtil(self.aws_access_key, self.aws_secret_key, region=self.aws_region)
+    self.fusion_monitor = FusionMonitorUtil(self.log, self.fusion_aws_util)
+    self.cp_monitor = FusionCPResourceMonitor(self.log, self.fusion_aws_util)
+    self.stop_run_event = threading.Event()
+```
+
+### Thread Coordination Pattern
+All long-running monitoring uses `threading.Event()` for clean lifecycle:
+```python
+# Start background monitoring
+cleanup_thread = threading.Thread(
+    target=self.cp_monitor.check_ebs_guest_vol_deletion,
+    kwargs={"tenant": tenant, "cluster": cluster, "stop_run_event": self.stop_run_event}
+)
+cleanup_thread.start()
+
+# In tearDown
+def tearDown(self):
+    self.stop_run_event.set()
+    for thread in self.background_threads:
+        thread.join()
+```
+
+### Delegation Pattern
+- Utility classes return booleans; test classes perform assertions
+- Monitoring logic belongs in Layer 2 utility classes, NOT in test classes
+- Use `FusionAWSUtil` for all AWS operations, never raw boto3
+
+## Key Constants
+- `FUSION_ACCELERATOR_IOPS = 16000` - Fusion accelerator instances use 16K IOPS volumes
+- `VBUCKET_COUNT = 128` - Fusion vBucket count
+- `DEFAULT_TIMEOUT = 1800` - Default monitoring timeout (30 minutes)
+- `EBS_CLEANUP_TIMEOUT = 1200` - EBS volume cleanup timeout (20 minutes)
+
+## Key Invariants to Validate
+1. **Accelerator Lifecycle**: Accelerator nodes appear during rebalance, get killed after completion
+2. **EBS Guest Volumes**: Created during rebalance, hydrated, cleaned up to 0 after completion
+3. **Cluster Health**: Returns to `healthy` state, no `deployment_failed`/`rebalance_failed`/`scaleFailed`
+4. **Fusion Status**: Remains `enabled` throughout operations
+5. **No CRITICAL Errors**: No CRITICAL in memcached logs, no core dumps, no hydration failures
+6. **ASG Cleanup**: Auto Scaling Groups cleaned up after rebalance
+
+## Test Execution
+```bash
+python testrunner.py -i node.ini -c conf/fusion_volume.conf \
+    -p aws_access_key=$AWS_ACCESS_KEY_ID,aws_secret_key=$AWS_SECRET_ACCESS_KEY \
+    -p region=us-east-1 -p h_scaling=True -p iterations=3
+```
+
+## Hard Constraints
+- All new test code goes in `pytests/aGoodDoctor/fusion/`
+- Follow the 3-layer architecture strictly
+- Never put monitoring logic directly in test classes
+- Use event-driven stop for all background threads
+- Use PrettyTable for structured logging (consistent with existing code)
+- Never hard-code AWS credentials or secrets
+- Proper cleanup in tearDown (stop events, thread joins, CloudTrail teardown)
+- Follow existing import patterns and class naming conventions
+
+## Reference Documentation
+- `pytests/aGoodDoctor/fusion/architecture.md` - Canonical architecture reference with runtime flows, threading model, and extensibility guidelines
+- `pytests/aGoodDoctor/fusion/README.md` - Quick start and API summaries
+- `pytests/aGoodDoctor/fusion/FIS-LIB-README.md` - FIS fallback testing details
+- `AGENTS.md` - Root TAF coding guidelines

.factory/settings.json

@@ -0,0 +1,5 @@
+{
+  "enabledPlugins": {
+    "core@factory-plugins": true
+  }
+}

.gitignore

@@ -38,9 +38,6 @@ build
 /bin/
 **/settings
 
-# Factory/AI tools
-**/.factory
-
 # Jython cache
 .jython_cache/
 
@@ -60,4 +57,4 @@ credentials.json
 # Node.js (for jscpd and other npm tools)
 node_modules/
 package-lock.json
-package.json
+package.json

connections/Rest_Connection.py

@@ -202,10 +202,13 @@ def urllib_request(self, api, method='GET', headers=None,
                        params={}, timeout=300, verify=False):
         session = requests.Session()
         headers = headers or self.get_headers_for_content_type_json()
-        params = json.dumps(params)
+        # For GET requests, params should be a dict for query parameters
+        # For other methods, JSON-encode as request body
+        if method != "GET":
+            params = json.dumps(params)
         try:
             if method == "GET":
-                resp = session.get(api, params=params, headers=headers,
+                resp = session.get(api, params=params if params else None, headers=headers,
                                    timeout=timeout, verify=verify)
             elif method == "POST":
                 resp = session.post(api, data=params, headers=headers,

couchbase_utils/bucket_utils/bucket_ready_functions.py

@@ -208,6 +208,7 @@ def _loader_dict(cluster, buckets, overRidePattern=None,
                     key_prefix = key_prefix or bucket.loadDefn.get("key_prefix", "test_docs-")
                     key_size = key_size or bucket.loadDefn.get("key_size", 20)
                     key_type = key_type or bucket.loadDefn.get("key_type", "SimpleKey")
+                    doc_size = workloads[i % len(workloads)].get("doc_size", 256)
                     model = model or bucket.loadDefn.get("model", "Hotel")
                     mockVector = mockVector or bucket.loadDefn.get("mockVector", False)
                     base64 = base64 or bucket.loadDefn.get("base64", False)
@@ -216,16 +217,20 @@ def _loader_dict(cluster, buckets, overRidePattern=None,
                         continue
                     if collection == "_default" and scope == "_default" and skip_default:
                         continue
-                    per_coll_ops = bucket.loadDefn.get("ops")//(len(bucket.scopes[scope].collections.keys()) - 1)
+                    if bucket.loadDefn.get("ops") and bucket.loadDefn.get("ops") not in [None, "None"]:
+                        per_coll_ops = bucket.loadDefn.get("ops")//(len(bucket.scopes[scope].collections.keys()) - 1)
+                    else:
+                        per_coll_ops = None
+                    JavaDocLoaderUtils.log.info(f"Loading {per_coll_ops} ops for {bucket.name+scope+collection}")
                     loader = SiriusCouchbaseLoader(
                         server_ip=cluster.master.ip, server_port=cluster.master.port,
                         username="Administrator", password="password",
                         bucket=bucket,
                         scope_name=scope, collection_name=collection,
-                        key_prefix=key_prefix, key_size=key_size, doc_size=256,
+                        key_prefix=key_prefix, key_size=key_size, doc_size=doc_size,
                         key_type=key_type, value_type=valType,
-                        create_percent=pattern["create"], read_percent=pattern["read"], update_percent=pattern["update"],
-                        delete_percent=pattern["delete"], expiry_percent=pattern["expiry"],
+                        create_percent=pattern.get("create", 0), read_percent=pattern.get("read", 0), update_percent=pattern.get("update", 0),
+                        delete_percent=pattern.get("delete", 0), expiry_percent=pattern.get("expiry", 0),
                         create_start_index=bucket.create_start , create_end_index=bucket.create_end,
                         read_start_index=bucket.read_start, read_end_index=bucket.read_end,
                         update_start_index=bucket.update_start, update_end_index=bucket.update_end,
@@ -276,17 +281,20 @@ def perform_load(cluster, buckets, wait_for_load=True,
                     if collection == "_default" and scope == "_default" and skip_default:
                         continue
                     loader = loader_map[bucket.name+scope+collection]
-                    loader.create_doc_load_task()
+                    result, json_response = loader.create_doc_load_task()
+                    if not result:
+                        JavaDocLoaderUtils.log.critical("Failed to create doc load task: %s" % json_response)
+                        return False
                     JavaDocLoaderUtils.doc_loading_tm.add_new_task(loader)
                     tasks.append(loader)
-
         if wait_for_load:
             JavaDocLoaderUtils.wait_for_doc_load_completion(cluster, tasks, wait_for_stats)
         else:
             return tasks
 
         if validate_data:
             JavaDocLoaderUtils.data_validation(cluster, skip_default=skip_default)
+        return []
 
     @staticmethod
     def load_sift_data(cluster=None, buckets=None, overRidePattern=None, skip_default=True,
@@ -318,6 +326,7 @@ def load_sift_data(cluster=None, buckets=None, overRidePattern=None, skip_defaul
                     key_prefix = bucket.loadDefn.get("key_prefix")
                     key_size = bucket.loadDefn.get("key_size")
                     key_type = bucket.loadDefn.get("key_type")
+                    doc_size = workload.get("doc_size", 256)
                     model = bucket.loadDefn.get("model")
                     mockVector = bucket.loadDefn.get("mockVector")
                     base64 = bucket.loadDefn.get("base64")
@@ -332,7 +341,7 @@ def load_sift_data(cluster=None, buckets=None, overRidePattern=None, skip_defaul
                         username="Administrator", password="password",
                         bucket=bucket,
                         scope_name=scope, collection_name=collection,
-                        key_prefix=key_prefix, key_size=key_size, doc_size=256,
+                        key_prefix=key_prefix, key_size=key_size, doc_size=doc_size,
                         key_type=key_type, value_type=valType,
                         create_percent=pattern["create"], read_percent=pattern["read"], update_percent=pattern["update"],
                         delete_percent=pattern["delete"], expiry_percent=pattern["expiry"],
@@ -380,7 +389,7 @@ def load_data(cluster, buckets=None, overRidePattern=None,
                                create_end=override_num_items or bucket.loadDefn.get("num_items"),
                                bucket=bucket)
 
-        JavaDocLoaderUtils.perform_load(cluster=cluster,
+        return JavaDocLoaderUtils.perform_load(cluster=cluster,
                             buckets=buckets,
                             overRidePattern=overRidePattern,
                             validate_data=validate_data,
@@ -389,21 +398,6 @@ def load_data(cluster, buckets=None, overRidePattern=None,
                             mutate=mutate,
                             suppress_error_table=suppress_error_table,
                             track_failures=track_failures)
-        if update:
-            for bucket in buckets:
-                JavaDocLoaderUtils.generate_docs(doc_ops=["update"],
-                                   update_start=0,
-                                   update_end=override_num_items or bucket.loadDefn.get("num_items"),
-                                   bucket=bucket)
-            JavaDocLoaderUtils.perform_load(cluster=cluster,
-                              buckets=buckets,
-                              overRidePattern={"create": 0, "read": 0, "update": 100, "delete": 0, "expiry": 0},
-                              validate_data=False,
-                              wait_for_load=wait_for_load,
-                              wait_for_stats=wait_for_stats,
-                              mutate=mutate,
-                              suppress_error_table=suppress_error_table,
-                              track_failures=track_failures)
 
 class DocLoaderUtils(object):
     log = logger.get("test")