Get Started

input:
  kafka:
    addresses: [${KAFKA_BROKERS}]
    topics: [usage-events]
    consumer_group: flexprice-collector

pipeline:
  processors:
    - mapping: |
        root.event_name = this.event_name
        root.external_customer_id = this.customer_id
        root.properties = this.properties.map_each(p -> p.value.string())
        root.timestamp = this.timestamp.or(now().format_timestamp("2006-01-02T15:04:05Z07:00"))

output:
  flexprice:
    api_host: https://api.flexprice.io
    api_key: ${FLEXPRICE_API_KEY}
    scheme: https
    batching:
      count: 10
      period: 2s

docker run -d \
  --name bento-collector \
  -e FLEXPRICE_API_KEY=fp_live_xxxxx \
  -e KAFKA_BROKERS=broker:9092 \
  -v $(pwd)/config.yaml:/bento.yaml:ro \
  ghcr.io/flexprice/bento-collector:latest \
  -c /bento.yaml

docker logs -f bento-collector

# Expected:
# INFO Flexprice output connected and ready
# INFO Input type kafka is now active

kubectl create secret generic bento-collector \
  --from-literal=FLEXPRICE_API_KEY=fp_live_xxxxx \
  --from-literal=KAFKA_BROKERS=broker:9092

apiVersion: apps/v1
kind: Deployment
metadata:
  name: bento-collector
spec:
  replicas: 2
  selector:
    matchLabels:
      app: bento-collector
  template:
    metadata:
      labels:
        app: bento-collector
    spec:
      containers:
      - name: bento
        image: ghcr.io/flexprice/bento-collector:latest
        args: ["-c", "/config/config.yaml"]
        envFrom:
        - secretRef:
            name: bento-collector
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"

output:
  flexprice:
    batching:
      count: 10      # Events per batch
      period: 2s     # Max wait time
    max_in_flight: 10  # Concurrent requests

input:
  kafka:
    fetch_buffer_cap: 256        # Fetch more messages for batching
    commit_period: 1s
    max_processing_period: 100ms

metrics:
  prometheus:
    enabled: true
    prefix: bento

http:
  enabled: true
  address: 0.0.0.0:4195