Logs

The Logs scraper queries log aggregation systems to extract configuration changes from log entries. It supports multiple log backends including Loki, GCP Cloud Logging, OpenSearch, BigQuery and Azure Log Analytics. This allows you to create configuration items and track changes based on log data.

Use Cases

Application Configuration Changes: Track config reloads and updates from application logs
Deployment Tracking: Monitor deployment events from CI/CD pipeline logs
Error Analysis: Create configuration items from error patterns in logs
Audit Trail: Track security and compliance events from audit logs
Performance Monitoring: Extract performance metrics as configuration changes

logs-scraper.yaml
apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: app-config-changes
  namespace: mc
spec:
  full: true
  logs:
    - id: None
      type: None
      loki:
        url: http://localhost:3100
        query: '{job="app"} |~ "Configuration reloaded:.*changed from.*to"'
        limit: '50'
        start: 24h
      transform:
        expr: |
          dyn(config.logs).map(line, {
            "changes": [
              {
                "external_change_id": line.hash,
                "change_type": "ConfigReload",
                "external_id": "fdee1b15-4579-499e-adc5-2817735ec3f6",
                "config_type": "Azure::AppRegistration",
                "created_at": line.firstObserved,
                "scraper_id": "all"
              }
            ]
          }).toJSON()

# curl -X POST http://localhost:3100/loki/api/v1/push \
#   -H "Content-Type: application/json" \
#   -d '{
#     "streams": [
#       {
#         "stream": {
#           "job": "app"
#         },
#         "values": [
#           ["'$(date +%s%N)'", "Configuration reloaded: database.max_connections changed from 100 to 200"],
#           ["'$(date +%s%N)'", "Configuration reloaded: server.timeout changed from 30s to 60s"],
#           ["'$(date +%s%N)'", "Configuration reloaded: cache.size changed from 1GB to 2GB"]
#         ]
#       }
#     ]
#   }'

Field	Description	Scheme	Required
`schedule`	Specify the interval to scrape in cron format. Defaults to every 60 minutes.	Cron
`retention`	Settings for retaining changes, analysis and scraped items	`Retention`
`logs`	Specifies the list of log configurations to scrape.	`[]Logs`	`true`

Logs

Field	Description	Scheme
`azureLogAnalytics`	Azure Log Analytics configuration	AzureLogAnalytics
`bigQuery`	BigQuery configuration for log scraping	BigQuery
`gcpCloudLogging`	GCP Cloud Logging configuration	GCPCloudLogging
`loki`	Loki configuration for log scraping	Loki
`openSearch`	OpenSearch configuration for log scraping	OpenSearch
`fieldMapping.dedupBy`	Fields to dedupe the returned logs by	`[]string`
`fieldMapping.groupBy`	Fields to group the returned logs by	`[]string`
`fieldMapping.host`	Source field names containing the host the log came from	`[]string`
`fieldMapping.id`	Source field names containing unique identifiers for deduplication. Common values: `id`, `event_id`, `trace_id`, `request_id`	`[]string`
`fieldMapping.ignore`	Fields to drop from the log labels	`[]string`
`fieldMapping.message`	Source field names containing the main log message content. Common values: `message`, `msg`, `log`, `text`, `body`	`[]string`
`fieldMapping.severity`	Source field names containing log level/severity. Used to categorize log entries. Common values: `level`, `severity`, `log_level`, `priority`	`[]string`
`fieldMapping.source`	Source field names containing the log source	`[]string`
`fieldMapping.timestamp`	Source field names containing timestamp data. The scraper tries each field in order until it finds a non-empty value. Common values: `@timestamp`, `timestamp`, `time`, `date`	`[]string`

Field Mapping

Different log systems use different field names for the same data. For example:

Timestamp: Elasticsearch uses @timestamp, Loki uses ts, CloudWatch uses timestamp
Message: Some systems use message, others use msg, log, or text
Severity: Could be level, severity, log_level, or priority

Field mapping normalizes these differences so your transform expressions work consistently regardless of the log source.

How It Works

When you specify multiple field names, the scraper tries each one in order and uses the first non-empty value:

fieldMapping:
  timestamp: ['@timestamp', 'timestamp', 'time']  # Try @timestamp first, then timestamp, then time
  message: ['message', 'msg', 'log']
  severity: ['level', 'severity']
  id: ['trace_id', 'request_id', 'event_id']

When to Use Field Mapping

Use field mapping when:

Your logs come from multiple sources with different schemas
You're migrating between log systems and field names changed
Third-party applications use non-standard field names
You want transforms to be portable across different backends

Example: Normalizing ELK and Loki logs

apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: unified-app-logs
spec:
  logs:
    # OpenSearch/Elasticsearch logs use @timestamp and message
    - openSearch:
        address: https://elasticsearch:9200
        index: app-logs-*
      fieldMapping:
        timestamp: ['@timestamp']
        message: ['message']
        severity: ['level', 'log.level']
        id: ['trace.id', 'request_id']

    # Loki logs use different field names
    - loki:
        url: http://loki:3100
        query: '{app="myservice"}'
      fieldMapping:
        timestamp: ['ts', 'timestamp']
        message: ['line', 'msg']
        severity: ['level', 'detected_level']
        id: ['traceID']

BigQuery

Field	Description	Scheme
`project*`	GCP project ID containing the BigQuery dataset	`string`
`query*`	SQL query to execute against BigQuery	`string`
`credentials`	GCP service account credentials	EnvVar

Example

bigquery-github-commits.yaml
apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: github-commits-logs
  namespace: mc
spec:
  full: true
  schedule: '@every 1h'
  logs:
    - name: golang-github-commits
      id: None
      type: None
      transform:
        expr: |
          dyn(config.logs).map(line, {
            "changes": [
              {
                "external_change_id": line.id,
                "change_type": "Commit",
                "external_id": "github://golang/go",
                "config_type": "GitHub::Repository",
                "created_at": line.firstObserved,
                "summary": line.message,
                "scraper_id": "all"
              }
            ]
          }).toJSON()
      bigQuery:
        project: workload-prod-eu-02
        query: |
          SELECT
            FORMAT_TIMESTAMP('%Y-%m-%dT%H:%M:%SZ', TIMESTAMP_SECONDS(committer.date.seconds)) as timestamp,
            CASE 
              WHEN REGEXP_CONTAINS(LOWER(message), r'fix|bug|error') THEN 'high'
              WHEN REGEXP_CONTAINS(LOWER(message), r'feat|add|new') THEN 'medium'
              ELSE 'info'
            END as severity,
            message,
            commit
          FROM `bigquery-public-data.github_repos.commits`
          Where 'golang/go' IN UNNEST(repo_name)
          ORDER BY committer.date.seconds DESC
          LIMIT 100
      fieldMapping:
        timestamp: ['timestamp']
        severity: ['severity']
        message: ['message']
        id: ['commit']

GCPCloudLogging

Field	Description	Scheme
`project*`	GCP project ID	`string`
`connection`	GCP connection to use for credentials	`string`
`credentials`	GCP service account credentials	EnvVar
`end`	End time for the query. Defaults to now	`string`
`filter`	Cloud Logging filter query	`string`
`limit`	Maximum number of log entries to return	`string`
`start`	Start time for the query. Supports datemath e.g. `now-24h`	`string`

Example

apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: gcp-audit-logs
spec:
  logs:
    - gcpCloudLogging:
        project: my-gcp-project
        filter: |
          protoPayload.serviceName="compute.googleapis.com"
          protoPayload.methodName:"compute.instances"
        start: now-24h
        limit: '100'
      transform:
        expr: |
          dyn(config.logs).map(line, {
            "changes": [{
              "change_type": "GCPResourceChange",
              "external_id": line.resource.labels.instance_id,
              "config_type": "GCP::Instance",
              "created_at": line.timestamp,
              "summary": line.protoPayload.methodName
            }]
          }).toJSON()

Loki

Field	Description	Scheme
`query*`	LogQL query to execute	`string`
`url*`	Loki server URL	`string`
`end`	End time for the query	`string`
`limit`	Maximum number of log entries to return	`string`
`password`	Basic auth password	EnvVar
`start`	Start time for the query (e.g., '24h', '7d')	`string`
`username`	Basic auth username	EnvVar

Example

loki-config-changes.yaml
apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: app-config-changes
  namespace: mc
spec:
  full: true
  logs:
    - id: None
      type: None
      loki:
        url: http://localhost:3100
        query: '{job="app"} |~ "Configuration reloaded:.*changed from.*to"'
        limit: '50'
        start: 24h
      transform:
        expr: |
          dyn(config.logs).map(line, {
            "changes": [
              {
                "external_change_id": line.hash,
                "change_type": "ConfigReload",
                "external_id": "fdee1b15-4579-499e-adc5-2817735ec3f6",
                "config_type": "Azure::AppRegistration",
                "created_at": line.firstObserved,
                "scraper_id": "all"
              }
            ]
          }).toJSON()

# curl -X POST http://localhost:3100/loki/api/v1/push \
#   -H "Content-Type: application/json" \
#   -d '{
#     "streams": [
#       {
#         "stream": {
#           "job": "app"
#         },
#         "values": [
#           ["'$(date +%s%N)'", "Configuration reloaded: database.max_connections changed from 100 to 200"],
#           ["'$(date +%s%N)'", "Configuration reloaded: server.timeout changed from 30s to 60s"],
#           ["'$(date +%s%N)'", "Configuration reloaded: cache.size changed from 1GB to 2GB"]
#         ]
#       }
#     ]
#   }'

AzureLogAnalytics

Field	Description	Scheme
`query*`	KQL (Kusto Query Language) query to execute	`string`
`workspaceID*`	Azure Log Analytics workspace ID to query	`string`
`clientID`	Azure client ID	EnvVar
`clientSecret`	Azure client secret	EnvVar
`connection`	Azure connection to use for credentials	`string`
`end`	End time for the query. Defaults to now	`string`
`limit`	Maximum number of log entries to return	`string`
`start`	Start time for the query. Supports datemath e.g. `now-24h`	`string`
`tenantID`	Azure tenant ID	`string`

Example

apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: azure-activity-logs
spec:
  logs:
    - azureLogAnalytics:
        connection: connection://azure/production
        workspaceID: 00000000-0000-0000-0000-000000000000
        start: now-24h
        query: |
          AzureActivity
          | where OperationNameValue startswith "MICROSOFT.COMPUTE"
          | order by TimeGenerated desc

OpenSearch

Field	Description	Scheme
`address*`	OpenSearch cluster address	`string`
`index*`	Index name or pattern	`string`
`query*`	OpenSearch query DSL	`string`
`connection`	Connection name for OpenSearch credentials	`string`
`limit`	Maximum number of documents to return	`string`
`password`	Basic auth password	EnvVar
`username`	Basic auth username	EnvVar

Example

apiVersion: configs.flanksource.com/v1
kind: ScrapeConfig
metadata:
  name: opensearch-security-events
spec:
  logs:
    - openSearch:
        address: https://opensearch-cluster:9200
        index: security-logs-*
        query: |
          {
            "query": {
              "bool": {
                "must": [
                  {"term": {"event_type": "authentication"}},
                  {"range": {"@timestamp": {"gte": "now-1h"}}}
                ]
              }
            }
          }
        limit: '1000'
        username:
          valueFrom:
            secretKeyRef:
              name: opensearch-creds
              key: username
        password:
          valueFrom:
            secretKeyRef:
              name: opensearch-creds
              key: password
      fieldMapping:
        timestamp: ['@timestamp', 'timestamp']
        message: ['message', 'event_description']
        severity: ['severity', 'log_level']
        id: ['event_id', 'transaction_id']

Use Cases​

Logs​

Field Mapping​

How It Works​

When to Use Field Mapping​

BigQuery​

GCPCloudLogging​

Loki​

AzureLogAnalytics​

OpenSearch​

Use Cases

Logs

Field Mapping

How It Works

When to Use Field Mapping

BigQuery

GCPCloudLogging

Loki

AzureLogAnalytics

OpenSearch