scgps/scgps-pg-sinker repository overview

⁠PostgreSQL Sinker Docker Implementation

This directory contains a Docker-based implementation of the SCGPS telemetry data sinker for PostgreSQL with PostGIS extension. The service consumes telemetry events from Kafka and stores them in PostgreSQL database with optimized geospatial support.

⁠Quick Start

1. Copy docker-compose.yaml-sample to docker-compose.yaml
   • You can also obtain the latest docker-compose.yaml-sample from the official repository: examples/pg_sinker_docker on GitLab⁠
2. Configure the environment variables in docker-compose.yaml
3. Run the service:

   docker-compose up -d
   

   Copy

The image can be pinned by tag or digest (for example scgps/scgps-pg-sinker:2.1.0) for deterministic deployments.

⁠Configuration Parameters

⁠Kafka Configuration

The Kafka connection uses SASL_SSL security protocol with SCRAM-SHA-512 mechanism. The client verifies the broker certificate and hostname on every supported endpoint. An invalid KAFKA_PROVIDER value causes an error before any connection is attempted. Provider selection (KAFKA_PROVIDER) is available on the Python consumer path only.

Recommended upgrade flow: upgrade the image first with unchanged environment variables (behavior unchanged), then switch endpoints later by updating KAFKA_BOOTSTRAP_SERVERS, KAFKA_SASL_PASS, and KAFKA_PROVIDER.

This sinker performs no consumer-group offset migration; transferring group positions between clusters is subject to separate coordination.

⁠Performance Tuning

⁠Event Filtering Options

You can filter events using one of the following methods (it is recommended to avoid using multiple filters simultaneously, but it is possible):

1. Drop List - Filter out specific event types:

   KAFKA_EVENT_DROP_LIST: "CustomerSpecEvent,FileStreamEvent"
   

   Copy

2. Keep List - Process only specific event types:

   KAFKA_EVENT_KEEP_LIST: "GPSPositionTelemetry,ExternalInputEvent"
   

   Copy

3. Plate Number Filter - Process events only from specific vehicles:

   KAFKA_EVENT_KEEP_PNUM_RID_LIST: "ABC-123,DEF-456"
   

   Copy

For a complete list of event types, refer to the default event parser configuration⁠.

⁠PostgreSQL Configuration

⁠Database Schema

The service requires two schemas in the database:

1. iot_telemetry - Raw telemetry events (INSERT only)

   • Optimized for high-speed data ingestion
   • Historical event preservation
   • PostGIS geospatial data types

2. iot_aggregates - Aggregated data (full CRUD operations)

   • Optimized for analytical processing
   • Continuous data aggregation
   • Geospatial query support

⁠Features

• SASL/SSL Kafka authentication
• Configurable batch processing
• Flexible event filtering
• Automatic database reconnection
• PostGIS geospatial extension support
• High-speed data ingestion optimization
• Native geospatial indexing and querying

⁠Requirements

• Docker and Docker Compose
• PostGIS extension (automatically included in postgis/postgis:15-3.3 Docker image)
• Network access to Kafka brokers
• Network access to PostgreSQL server
• Python packages (automatically installed in Docker):
  • psycopg2-binary
  • sqlalchemy
  • geoalchemy2
  • confluent-kafka
  • protobuf==3.14.0 (installed with --no-binary=protobuf)

⁠Troubleshooting

1. Database Connection Issues

   • Verify PostgreSQL server is accessible
   • Check PostGIS extension installation
   • Verify credentials and database existence
   • Check database logs for connection errors

2. Kafka Connection Issues

   • Verify Kafka credentials
   • Check network connectivity to Kafka brokers
   • Verify topic permissions

3. Performance Issues

   • Adjust KAFKA_BATCH_SIZE and KAFKA_MAX_WAIT_IN_SEC
   • Monitor thread pool utilization
   • Check PostgreSQL configuration (connections, memory, etc.)
   • Verify PostGIS index usage

⁠Support

For additional support or questions about:

• Kafka credentials and topics: Contact SCGPS support
• Implementation details: Refer to the main documentation⁠
• Database setup: Consult your database administrator