NMS-20032: implement a postgresql-based Flows feature

[dino2gnt]

This implements a PostgreSQL-based flows feature, mimicking the functionality of and implementing the same features as features/flows/elastic, including a PostgresFlowRepository and PostgresFlowQueryService. The PostgresFlowQueryService bakes in the proportional sum aggregation currently packaged for Elasticseach as Drift.

It supports both "internal" and "external" databases and the feature is loadable on Sentinel, targeting an external database for offload.

The table schema is fairly simple and uses PostgreSQL native partitioning:

                        Partitioned table "public.flow"
      Column       |           Type           | Collation | Nullable | Default
-------------------+--------------------------+-----------+----------+---------
 flow_ts           | timestamp with time zone |           | not null |
 received_at       | timestamp with time zone |           | not null | now()
 delta_switched    | bigint                   |           | not null |
 last_switched     | bigint                   |           | not null |
 first_switched    | bigint                   |           |          |
 bytes             | bigint                   |           |          |
 packets           | bigint                   |           |          |
 sampling_interval | double precision         |           |          |
 direction         | text                     |           |          |
 application       | text                     |           |          |
 convo_key         | text                     |           |          |
 src_addr          | inet                     |           |          |
 dst_addr          | inet                     |           |          |
 protocol          | smallint                 |           |          |
 dscp              | smallint                 |           |          |
 exporter_node_id  | integer                  |           |          |
 input_snmp        | integer                  |           |          |
 output_snmp       | integer                  |           |          |
 location          | text                     |           |          |
 document          | jsonb                    |           | not null |
Partition key: RANGE (flow_ts)
Indexes:
    "flow_application_idx" btree (application)
    "flow_convo_key_idx" btree (convo_key)
    "flow_document_gin_idx" gin (document)
    "flow_exporter_idx" btree (exporter_node_id)
    "flow_ts_idx" btree (flow_ts)

On a Core instance, partition creation and expiration is managed by vacuumd. On an external database, pg_partman and pg_cron are recommended.

In my testing, I was able to scale this to ~15,000 flow documents a second with only the tunables suggested by pgtune.leopard.in.ua.

Testing environment consisted of:
10c / 16G PostgreSQL 18:

# from pgtune for OLTP
max_connections = 100
shared_buffers = 4GB
effective_cache_size = 12GB
maintenance_work_mem = 1GB
checkpoint_completion_target = 0.9
wal_buffers = 16MB
default_statistics_target = 100
random_page_cost = 1.1
effective_io_concurrency = 1000
work_mem = 34952kB
huge_pages = try
jit = off
wal_compression = lz4
min_wal_size = 2GB
max_wal_size = 8GB
max_worker_processes = 8
max_parallel_workers_per_gather = 4
max_parallel_workers = 8
max_parallel_maintenance_workers = 4

6c / 6g OpenNMS

• Telemetryd:

       <parser name="Netflow-9-Parser" class-name="org.opennms.netmgt.telemetry.protocols.netflow.parser.Netflow9UdpParser" queue="Netflow-9">
               <parameter key="threads" value="24"/>
               <parameter key="dnsLookupsEnabled" value="false"/>
       </parser>
       
 <queue name="Netflow-9" threads="24" batch-size="10000" queue-size="100000">

• org.opennms.features.flows.persistence.postgres.cfg

datasource.url=jdbc:postgresql://flowdb.test.example.com:5432/flowdb
datasource.username=flow
datasource.password=flow
datasource.databaseName=flowdb
runSchemaChangelog=false

batchSize=12000
flushIntervalMs=5000
queueCapacity=100000
queryThreads=4
writerThreads=10

• Kernel tunables:

net.core.rmem_default = 67108864
net.core.rmem_max = 67108864

There is some OSGi magic here to allow the PostgreSQL feature to supersede the Elasticsearch feature so the two don't load together that I don't claim to understand, I wanted to be able to specify the opennms-flows-postgres feature only and not have the two features conflict.

Assisted by Anthropic Claude Opus 4.8

External References

• Jira (Issue Tracker): https://opennms.atlassian.net/browse/NMS-20032

[marshallmassengill]

I gave some feedback which Dino implemented over the weekend. Overall I think this is good but it's probably worth a discussion before we merge it just to make sure we've covered our bases.