Developing Collectors
Add support for a new AWS service by creating a collector class.
BaseCollector ABC
from sofe.collectors.aws.base import BaseCollector
from sofe.models import Resource
class MyServiceCollector(BaseCollector):
resource_type = "aws.myservice" # BYaML-compatible type
def collect(self) -> list[Resource]:
try:
client = self.session.client('myservice', region_name=self.region)
items = client.list_things().get('Things', [])
return [self._make_resource(
resource_id=item['Id'],
tags=item.get('Tags', {}),
properties={'status': item.get('Status')},
metrics={'monthly_cost': 0}, # fetch from Cost Explorer
) for item in items]
except Exception as e:
print(f" ⚠️ MyService scan failed: {e}")
return []Steps to Add a Collector
- 1. Create file:
sofe/collectors/aws/myservice.py - 2. Implement
collect()→ returns list of Resource objects - 3. Register in
sofe/collectors/aws/__init__.py→ add to COLLECTORS dict - 4. Test:
python -c "from sofe.collectors.aws import COLLECTORS; print(len(COLLECTORS))" - 5. Write a policy that uses your collector's metrics
- 6. Submit PR to
github.com/breakingthecloud/sofe
Resource Model
@dataclass
class Resource:
resource_id: str # unique ID (instance ID, bucket name, etc.)
resource_type: str # "aws.ec2", "aws.s3", etc.
region: str # "us-east-1" or "global"
account_id: str # AWS account ID
tags: dict = field(default_factory=dict)
properties: dict = field(default_factory=dict)
metrics: dict = field(default_factory=dict) # what policies evaluate againstTips
- → Use
self._make_resource()helper — pre-fills region, account, type - → Always wrap in try/except — permission errors shouldn't crash the scan
- → Add
has_tag:*metrics in the enrichment step (already done automatically) - → Metrics dict keys should match what policies reference in their
rule.metric