Support for stream processing tools such as Kafka and NSQ

CHANGELOG.md

@@ -6,188 +6,252 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+
 ### Added
-- Option to configure trusted_proxies for which the X-Forwarded-For and X-Real-IP headers are used as client IP (b062edca)
+
+### Changed
+
+### Removed
+
+### Fixed
+
+## [1.10.0] - 2026-07-xx
+
+### Added
+
+- Support for stream processing tools like kafka or nsq
+- Option to configure trusted_proxies for which the X-Forwarded-For and X-Real-IP headers are used as client IP (b062edc)
+- Add `excluded_logs` config option to skip specific logs and operators. Works both for classic and tiled logs. (170fbeb)
 - Added multiple unit tests for ensuring correct functionality
-- Add `excluded_logs` config option to skip specific logs and operators. Works both for classic and tiled logs.
 
 ### Changed
-- Updated weak cipher suites to stronger ones (939517cd)
+
+- Updated weak cipher suites to stronger ones (939517c)
 - Updated http client settings to prevent timeouts and other connectivity issues
 - Updated http server settings to allow for higher delays
-- Minor code improvements and refactoring, mostly style related
 - Updated batch size from 100 to 256 (#97)
 
-### Removed
-
 ### Fixed
-- Calculate correct starting tile index for static ct logs
-- Use proper websocket close code (1008) instead of 1005, which wasn't sent to the client
-- Respect ct_index path in config file for the `create-index` command
-- Ensure all additionalLog entries are processed correctly
 
-### Docs
+- Prevent excessive tile fetching for static CT logs (#104)
+- Calculate correct starting tile index for static ct logs (a2b4a9f)
+- Use proper websocket close code (1008) instead of 1005, which wasn't sent to the client (76ff240)
+- Respect ct_index path in config file for the `create-index` command (67838a9)
+- Ensure all additionalLog entries are processed correctly (c8bd735)
 
 ## [1.9.0] - 2026-04-03
+
 ### Added
+
 - Ability to store and resume processing of certs from where it left off after a restart - see sample config "recovery" (#49)
 - New CLI switch for creating an index file from a CT log (#49)
-- Support for [Static CT](https://github.com/C2SP/C2SP/blob/main/static-ct-api.md) logs 
+- Support for [Static CT](https://github.com/C2SP/C2SP/blob/main/static-ct-api.md) logs
 - Check for retired CT logs and prevent them from being watched / stop watching them (#77)
 - Accept websocket connections from all origins
 - Option to disable the default logs provided by Google - see sample config "disable_default_logs"
 - Use of cobra for CLI argument parsing. New commands for displaying version and creating an index file
-- Override confiuration options via environment variables (e.g. `CERTSTREAM_WEBSERVER_LISTEN_PORT=1234` to change the listen port)
+- Override configuration options via environment variables (e.g. `CERTSTREAM_WEBSERVER_LISTEN_PORT=1234` to change the listen port)
 - New field in json `source.timestamp` representing the timestamp of when the cert was processed by the CT log (#98)
 
 ### Changed
+
 - **Breaking:** The configuration file for the docker container is now read from the /app/config/ directory (b9e5e6)
 
 ### Removed
+
 - Deleted non-functional Dodo log from sample config (#78)
 
 ### Fixed
+
 - Properly remove stopped ct log workers (#74)
 - Added missing fields certificatePolicies and ctlPoisonByte (#85)
 - Prevent race condition caused by simultaneous rw access to logmetrics (#91)
 - Properly display metrics for all initially watched logs (#95)
 - Properly add new metrics for all newly found logs (#96)
 
 ## [1.8.2] - 2025-11-22
+
 ### Fixed
+
 - Added missing fields certificatePolicies and ctlPoisonByte (#85)
 
 ## [1.8.1] - 2025-05-04
+
 ### Fixed
+
 - No longer reject URLs with trailing slashes defined in the `additional_logs` config (#62)
 - When using `drop_old_logs` in the config, the server won't remove logs defined in `additional_logs` anymore (#64)
 
 ## [1.8.0] - 2025-05-03
+
 ### Security
+
 - Close several CVEs in x/crypto and x/net dependencies (#59)
 
 ### Added
+
 - New CLI tool for fetching certificates from a CT log (#47)
 - Ability to add custom CT logs to the config (#56)
 - Remove old CT logs as soon as they are removed from the Google CT Loglist (#60)
 - New configuration for buffer sizes (#58)
 
 ### Fixed
+
 - Properly handle IPv6 addresses in config (#61)
 
 ## [1.7.1] - 2025-05-03
+
 ### Fixed
+
 - Properly handle IPv6 addresses in config (#61)
 
 ## [1.7.0] - 2024-08-20
+
 ### Added
+
 - Support for websocket compression - disabled by default (#40)
 - Support for non-browsers by implementing server initiated heartbeats (#39)
 - Start new ct-watchers as new ct logs become available (#42)
 - More logging to document currently watched logs (03d878e)
- 
+
 ### Changed
+
 - Changed log output to be better grepable (5c055cc)
 - Update ct log update interval to once per hour instead of once per 6 hours as previously (9b6e77d)
 
 ### Fixed
+
 - Fixed a possible race condition when accessing metrics
 
 ## [1.6.0] - 2024-03-05
+
 ### Added
+
 - New metric for skipped certs per client (#34)
 
 ## [1.5.2] - 2024-02-17
+
 ### Fixed
+
 - Fixed an issue with ip whitelists for the websocket server (#33)
 
 ## [1.5.1] - 2024-01-18
+
 ### Fixed
-- Fixed a rare issue where it was possible for the all_domains json property (or data property in case of the domains-only endpoint) to be null 
+
+- Fixed a rare issue where it was possible for the all_domains json property (or data property in case of the domains-only endpoint) to be null
 
 ## [1.5.0] - 2023-12-21
+
 ### Added
+
 - New `-version` switch to print version and exit afterwards
 - Print version on every run of the tool
 - Count and log number of skipped certificates per client
 
 ### Changed
+
 - Update to chi/v5
 - Update ct-watcher timeout from 5 to 30 seconds
 
 ### Fixed
+
 - Prevent invalid subscription types to be used
 - Kill connection after broadcasthandler was stopped
 
 ## [1.4.0] - 2023-11-29
+
 ### Added
+
 - Config option to use X-Forwarded-For or X-Real-IP header as client IP
 - Config option to whitelist client IPs for both websocket and metrics endpoints
 - Config option to enable system metrics (cpu, memory, etc.)
 
 ## [1.3.2] - 2023-11-28
+
 ### Fixed
+
 - Memory leak related to clients disconnecting from the websocket not being handled properly
 
 ## [1.3.1] - 2023-09-18
+
 ### Changed
+
 - Updated config.sample.yaml to run both certstream and prometheus metrics on same socket
 
 ### Docs
+
 - Fixed wrong docker command in readme
 
 ## [1.3.0] - 2023-04-11
+
 ### Added
+
 - Calculate and display Sha256 sum of certificate
 
 ### Changed
+
 - Update dependencies
 - Better logging for CT log errors
 
 ### Fixed
+
 - End execution after all workers stopped
 - Implement timeout for the http client
 - Keep ct watcher from crashing upon a connection reset from server
 
 ## [1.2.2] - 2023-01-10
+
 ### Added
+
 - Two docker-compose files
 - Check for presence of .yml or .yaml files in the current directory
 
 ### Fixed
+
 - Handle sudden disconnects of CT logs
 
 ### Docs
-- Added [wiki entry for docker-compose](https://github.com/d-Rickyy-b/certstream-server-go/wiki/Collecting-and-Visualizing-Metrics) 
+
+- Added [wiki entry for docker-compose](https://github.com/d-Rickyy-b/certstream-server-go/wiki/Collecting-and-Visualizing-Metrics)
 
 ## [1.2.1] - 2022-12-16
+
 ### Changed
+
 - Updated ci pipeline to use new setup-go and checkout actions
 - Use correct package name `github.com/d-Rickyy-b/certstream-server-go`
 
 ## [1.2.0] - 2022-12-15
+
 ### Added
+
 - Log x-Forwarded-For header for requests
 - More logging for certain error situations
 - Add operator to ct log cert count metrics
 
 ### Changed
+
 - Updated certificate-transparency-go dependency to v1.1.4
 - Code improvements, adhering to styleguide
 - Rename module to certstream-server-go
 - Use log_list.json instead of all_logs_list.json
 
 ## [1.1.0] - 2022-10-19
+
 Fix for missing loglist urls.
 
 ### Fixed
+
 Fixed the connection issue due to the offline Google loglist urls.
 
 ## [1.0.0] - 2022-08-08
+
 Initial release! First stable version of certstream-server-go is published as v1.0.0
 
-[unreleased]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.9.0...HEAD
-[1.8.2]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.2...v1.9.0
+[unreleased]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.10.0...HEAD
+[1.10.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.9.0...v1.10.0
+[1.9.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.2...v1.9.0
 [1.8.2]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.1...v1.8.2
 [1.8.1]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.0...v1.8.1
 [1.8.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.7.1...v1.8.0

cmd/certpicker/main.go

@@ -65,7 +65,7 @@ func main() {
 		log.Fatalln("Error getting entry from CT log: ", getEntryErr)
 	}
 
-	// Loop over entries and pars each one.
+	// Loop over entries and parse each one.
 	for _, leafEntry := range entries.Entries {
 		rawLogEntry, err := ct.RawLogEntryFromLeaf(certID, &leafEntry)
 		if err != nil {

config.sample.yaml

@@ -37,6 +37,35 @@ prometheus:
   whitelist:
     - "127.0.0.1/8"
 
+# Configuration related to external stream processing tools go here.
+stream_processing:
+  # List of stream processing configurations.
+  # Each configuration specifies a stream processing tool to use and its settings.
+  - name: "kafka-prod"
+    # Type describes the type of stream processing tool to use.
+    # Supported types: "kafka", "nqs"
+    type: "kafka"
+    # The stream type can be any of "full", "lite", "domains-only" - same as the certstream websocket urls
+    # Default is "full"
+    stream: "full"
+    # Enable or disable this stream processing configuration.
+    enabled: true
+    # The server address and port to connect to the stream processing tool
+    server_addr: "127.0.0.1"
+    server_port: 9092
+    topic: "certstream"
+    # Available compression types depend on the stream processing tool.
+    # E.g. for Kafka, supported types are "none" / "", "gzip", "snappy", "lz4", "zstd"
+    compression: ""
+
+  - name: "nqs-prod"
+    type: "nqs"
+    stream: "lite"
+    enabled: false
+    server_addr: "127.0.0.1"
+    server_port: 9092
+    topic: "certstream"
+
 general:
   # DisableDefaultLogs indicates whether the default logs used in Google Chrome and provided by Google should be disabled.
   disable_default_logs: false
@@ -66,9 +95,8 @@ general:
     websocket: 300
     # Buffer for each CT log connection
     ctlog: 1000
-    # Combined buffer for the broadcast manager
-    broadcastmanager: 10000
-
+    # Combined buffer for the cert dispatcher
+    dispatcher: 10000
   # Google regularly updates the log list. If this option is set to true, the server will remove all logs no longer listed in the Google log list.
   # This option defaults to true. See https://github.com/d-Rickyy-b/certstream-server-go/issues/51
   drop_old_logs: true

docker/docker-compose.metrics.yml

@@ -1,6 +1,6 @@
 version: '2'
 
-# Make sure to create the sub directories "prometheus", "prometheus_data", "grafana", "grafana_data" and "certstream"
+# Make sure to create the subdirectories "prometheus", "prometheus_data", "grafana", "grafana_data" and "certstream"
 # and create the config files for all three services. For further details please refer to https://github.com/d-Rickyy-b/certstream-server-go/wiki/Collecting-and-Visualizing-Metrics
 
 networks:
@@ -30,7 +30,7 @@ services:
     ports:
       # Exposing Prometheus is NOT required, if you don't want to access it from outside the Docker network.
       # Using localhost enables you to use a reverse proxy (e.g. with basic auth) to access Prometheus in a more secure way.
-      - 127.0.0.1:9090:9090
+      - "127.0.0.1:9090:9090"
     networks:
       - monitoring
     extra_hosts:
@@ -44,7 +44,7 @@ services:
     depends_on:
       - prometheus
     ports:
-      - 127.0.0.1:8082:3000
+      - "127.0.0.1:8082:3000"
     volumes:
       - ./grafana_data:/var/lib/grafana
       - ./grafana/provisioning/:/etc/grafana/provisioning/
@@ -60,7 +60,7 @@ services:
     # Configure the service to run as specific user.
     # user: "1000:1000"
     ports:
-      - 127.0.0.1:8080:80
+      - "127.0.0.1:8080:80"
       # Don't forget to open the other port in case you run the Prometheus endpoint on another port than the websocket server.
       # - 127.0.0.1:8081:81
     volumes:

go.mod

@@ -8,6 +8,8 @@ require (
 	github.com/google/certificate-transparency-go v1.3.3
 	github.com/google/trillian v1.7.3
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
+	github.com/nsqio/go-nsq v1.1.0
+	github.com/segmentio/kafka-go v0.4.51
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
 	golang.org/x/crypto v0.49.0
@@ -17,8 +19,11 @@ require (
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
+	github.com/golang/snappy v0.0.1 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/klauspost/compress v1.15.9 // indirect
 	github.com/pelletier/go-toml/v2 v2.3.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 	github.com/sagikazarmark/locafero v0.12.0 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
 	github.com/spf13/cast v1.10.0 // indirect