Just spent my weekend elbows-deep in Python and the Zscaler and LogRhythm APIs, and I have to say, the feeling of seeing those first firewall logs pop into our LogRhythm dashboard was worth every minute. Our security team has been wanting to get Zscaler Internet Access logs ingested for better user-risk correlation, but the official vendor module quote was... steep. We're talking a $5k initial cost, plus annual maintenance.
So I did what any DevOps engineer with a slight aversion to unnecessary spend would do: I built a connector. The core logic isn't too complex, but there are a few nuances around pagination, checkpointing, and log formatting that took some time to get right. I figured I'd share the architecture and some key snippets, in case anyone else is facing a similar "build vs. buy" decision for a cloud proxy log source.
My approach uses a lightweight container running on our internal K8s cluster. It's essentially a scheduled Python job that:
1. Pulls logs from the Zscaler Cloud Audit API (`/api/v1/auditLogs`) and Web Logs API.
2. Transforms the JSON into the specific key-value pairs LogRhythm's API expects.
3. Batches and forwards them using LogRhythm's Data API (`/lr-drilldown-api/record`).
4. Maintains a small state file (stored in a PVC) to track the last successful timestamp, ensuring we never miss logs on a pod restart.
Here's the heart of the transformation logic for the audit logs. The key was mapping Zscaler's field names to LogRhythm's expected schema.
```python
def transform_to_lr_schema(zscaler_record, log_type):
lr_record = {
'originEntityId': OUR_ORIGIN_ID,
'originHostId': OUR_HOST_ID,
'message': f"Zscaler {log_type}: {zscaler_record.get('action', 'N/A')}",
'logSource': 'Zscaler-Connector',
'timestamp': zscaler_record.get('timestamp')
}
# Map specific fields for meaningful drill-down
if log_type == 'audit':
lr_record['subjectUser'] = zscaler_record.get('user')
lr_record['object'] = zscaler_record.get('resource')
lr_record['action'] = zscaler_record.get('action')
lr_record['reason'] = zscaler_record.get('clientIP')
return {k: v for k, v in lr_record.items() if v is not None}
```
The real savings came from avoiding the module cost, but the hidden benefits are maybe even better:
* **No Vendor Lock-in:** We own the code. If we need to add a new log type or change a field, we just do it.
* **GitOps Friendly:** The entire solution is defined in a Git repo - Dockerfile, Python script, K8s CronJob manifest. Changes are peer-reviewed and deployed via our existing CI/CD pipeline.
* **Cost Control:** We can fine-tune the polling frequency and log volume we pull, which directly impacts our LogRhythm data ingestion costs.
The total build time was about 16 hours spread over Saturday and Sunday, including testing and writing the deployment manifests. That's a pretty fantastic ROI when you consider the alternative. It also gives us a reusable pattern for other cloud services that charge exorbitant fees for "official" log connectors.
Has anyone else gone down a similar path for Zscaler or other SaaS platforms? I'm curious about how you handled things like error backoff or high-volume log bursts. I'm thinking of adding a small Redis queue for resilience next.
— francesc
— francesc