Produce updates to kafka

.gitignore

@@ -7,3 +7,4 @@
 docs/sphinx/_build/
 nipap-www/nipap_www.egg-info
 nipap/nipap.egg-info
+/.idea/

nipap/nipap.conf.dist

@@ -212,3 +212,41 @@ secret_key = {{WWW_SECRET_KEY}}
 # otlp_http_endpoint=http://opentelemetry-collector:4318/v1/traces
 # Set sampler. Valid values are always_on, always_off, parentbased_always_on, parentbased_always_off, traceidratio and parentbased_traceidratio. Default is parentbased_always_on.
 # otel_traces_sampler = always_on
+
+#
+# Kafka event producer configuration
+#
+[kafka]
+# Enable running the external kafka producer process (true/false)
+# If true, nipapd will spawn the kafka_producer as a separate process.
+enabled = false
+
+# Comma-separated list of Kafka brokers, e.g. localhost:9092,broker2:9092
+#brokers = localhost:9092
+
+# Poll interval in seconds for the kafka producer to poll the DB
+#poll_interval = 2
+
+# Topic prefix for produced events (defaults to "nipap.")
+#topic_prefix = nipap.
+
+# Security protocol for Kafka connection (e.g., PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL)
+# Defaults to PLAINTEXT if not specified
+#security_protocol = PLAINTEXT
+
+# SASL authentication settings (required when using SASL_PLAINTEXT or SASL_SSL)
+# SASL mechanism to use (e.g., PLAIN, SCRAM-SHA-256, SCRAM-SHA-512)
+#sasl_mechanism = PLAIN
+
+# SASL username for authentication
+#sasl_username = your_username
+
+# SASL password for authentication
+#sasl_password = your_password
+
+# SSL/TLS settings (used with SSL or SASL_SSL security protocols)
+# Path to CA certificate file for SSL/TLS certificate verification
+#ssl_cafile = /path/to/ca-cert.pem
+
+# Enable/disable SSL hostname verification (true/false)
+#ssl_check_hostname = true

nipap/nipap/__init__.py

@@ -1,5 +1,5 @@
 __version__		= "0.32.7"
-__db_version__	= 7
+__db_version__	= 8
 __author__		= "Kristian Larsson, Lukas Garberg"
 __author_email__ = "[email protected], [email protected]"
 __copyright__	= "Copyright 2011-2014, Kristian Larsson, Lukas Garberg"

nipap/nipap/db_schema.py

@@ -529,7 +529,17 @@
 	RETURN (part_one::bigint << 32) + part_two::bigint;
 END;
 $_$ LANGUAGE plpgsql IMMUTABLE STRICT;
-"""
+
+CREATE OR REPLACE FUNCTION tf_kafka_produce_event() RETURNS trigger AS $$
+BEGIN
+	IF TG_OP = 'DELETE' THEN
+		INSERT INTO kafka_produce_event (table_name, event_type, payload) VALUES (TG_TABLE_NAME, TG_OP, row_to_json(OLD)::jsonb);
+	ELSIF OLD IS DISTINCT FROM NEW THEN
+		INSERT INTO kafka_produce_event (table_name, event_type, payload) VALUES (TG_TABLE_NAME, TG_OP, row_to_json(NEW)::jsonb);
+	END IF;
+	RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;"""
 
 ip_net = """
 --------------------------------------------
@@ -538,7 +548,7 @@
 --
 --------------------------------------------
 
-COMMENT ON DATABASE %s IS 'NIPAP database - schema version: 7';
+COMMENT ON DATABASE %s IS 'NIPAP database - schema version: 8';
 
 CREATE EXTENSION IF NOT EXISTS ip4r;
 CREATE EXTENSION IF NOT EXISTS hstore;
@@ -790,7 +800,21 @@
 CREATE INDEX ip_net_log__prefix__index ON ip_net_log(prefix_id);
 CREATE INDEX ip_net_log__pool__index ON ip_net_log(pool_id);
 
-"""
+--
+-- Kafka event table and triggers
+--
+-- This table is used as a queue for the external kafka_producer process.
+-- Triggers on the core tables insert events here. The daemon will enable or
+-- disable these triggers at startup depending on configuration.
+--
+CREATE TABLE IF NOT EXISTS kafka_produce_event (
+	id SERIAL PRIMARY KEY,
+	table_name TEXT NOT NULL,
+	event_type TEXT NOT NULL,
+	payload JSONB,
+	processed BOOLEAN DEFAULT FALSE,
+	created_at TIMESTAMP WITH TIME ZONE DEFAULT now()
+);"""
 
 triggers = """
 --
@@ -1768,7 +1792,25 @@
 	WHEN (OLD.ipv4_default_prefix_length IS DISTINCT FROM NEW.ipv4_default_prefix_length
 		OR OLD.ipv6_default_prefix_length IS DISTINCT FROM NEW.ipv6_default_prefix_length)
 	EXECUTE PROCEDURE tf_ip_net_pool__iu_before();
-"""
+
+-- Triggers that write to kafka_produce_event
+CREATE TRIGGER trigger_kafka_ip_net_plan
+	AFTER INSERT OR UPDATE OR DELETE
+	ON ip_net_plan
+	FOR EACH ROW
+	EXECUTE PROCEDURE tf_kafka_produce_event();
+
+CREATE TRIGGER trigger_kafka_ip_net_vrf
+	AFTER INSERT OR UPDATE OR DELETE
+	ON ip_net_vrf
+	FOR EACH ROW
+	EXECUTE PROCEDURE tf_kafka_produce_event();
+
+CREATE TRIGGER trigger_kafka_ip_net_pool
+	AFTER INSERT OR UPDATE OR DELETE
+	ON ip_net_pool
+	FOR EACH ROW
+	EXECUTE PROCEDURE tf_kafka_produce_event();"""
 
 upgrade = [
 """
@@ -2272,4 +2314,59 @@
 -- update database schema version
 COMMENT ON DATABASE %s IS 'NIPAP database - schema version: 7';
 """,
+"""
+--
+-- Upgrade from NIPAP database schema version 7 to 8
+--
+
+--
+-- Kafka event table and triggers
+--
+-- This table is used as a queue for the external kafka_producer process.
+-- Triggers on the core tables insert events here. The daemon will enable or
+-- disable these triggers at startup depending on configuration.
+--
+CREATE TABLE IF NOT EXISTS kafka_produce_event (
+	id SERIAL PRIMARY KEY,
+	table_name TEXT NOT NULL,
+	event_type TEXT NOT NULL,
+	payload JSONB,
+	processed BOOLEAN DEFAULT FALSE,
+	created_at TIMESTAMP WITH TIME ZONE DEFAULT now()
+);
+
+CREATE OR REPLACE FUNCTION tf_kafka_produce_event() RETURNS trigger AS $$
+BEGIN
+	IF TG_OP = 'DELETE' THEN
+		INSERT INTO kafka_produce_event (table_name, event_type, payload) VALUES (TG_TABLE_NAME, TG_OP, row_to_json(OLD)::jsonb);
+	ELSIF OLD IS DISTINCT FROM NEW THEN
+		INSERT INTO kafka_produce_event (table_name, event_type, payload) VALUES (TG_TABLE_NAME, TG_OP, row_to_json(NEW)::jsonb);
+	END IF;
+	RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Triggers that write to kafka_produce_event
+CREATE TRIGGER trigger_kafka_ip_net_plan
+	AFTER INSERT OR UPDATE OR DELETE
+	ON ip_net_plan
+	FOR EACH ROW
+	EXECUTE PROCEDURE tf_kafka_produce_event();
+
+CREATE TRIGGER trigger_kafka_ip_net_vrf
+	AFTER INSERT OR UPDATE OR DELETE
+	ON ip_net_vrf
+	FOR EACH ROW
+	EXECUTE PROCEDURE tf_kafka_produce_event();
+
+CREATE TRIGGER trigger_kafka_ip_net_pool
+	AFTER INSERT OR UPDATE OR DELETE
+	ON ip_net_pool
+	FOR EACH ROW
+	EXECUTE PROCEDURE tf_kafka_produce_event();
+
+
+-- update database schema version
+COMMENT ON DATABASE %s IS 'NIPAP database - schema version: 8';
+""",
 ]