Going full reactive with Kafka; ditching PostgreSQL+TimescaleDB

Author: only-dev-ops

docker-compose.yml

@@ -16,18 +16,6 @@ services:
       - ./prometheus.yml:/etc/prometheus/prometheus.yml
       - prometheus-data:/prometheus
 
-  timescaledb:
-    image: timescale/timescaledb:latest-pg15
-    ports:
-      - "5432:5432"
-    environment:
-      POSTGRES_USER: nhp
-      POSTGRES_PASSWORD: nhppass
-      POSTGRES_DB: nhpdb
-    volumes:
-      - timescale-data:/var/lib/postgresql/data
-
-
   grafana:
     image: grafana/grafana
     ports:
@@ -42,6 +30,5 @@ services:
 
 
 volumes:
-  timescale-data:
   prometheus-data:
   grafana-data:

src/main/java/com/nhp/config/StreamProperties.java

@@ -12,8 +12,6 @@
 
 import com.nhp.dto.MultipleMonitoredPrefixesRequest;
 import com.nhp.dto.SingleMonitoredPrefixRequest;
-import com.nhp.model.MonitoredPrefix;
-import com.nhp.repos.MonitoredPrefixesRepo;
 import com.nhp.stream.RipeStreamClient;
 
 import java.time.Instant;
@@ -23,51 +21,43 @@
 @RequestMapping("/api/prefixes")
 public class MonitoredPrefixController {
 
-    @Autowired
-    private final MonitoredPrefixesRepo monitoredPrefixesRepo;
-
     @Autowired
     private final RipeStreamClient ripeStreamClient;
 
-    public MonitoredPrefixController(MonitoredPrefixesRepo repo, RipeStreamClient client) {
-        monitoredPrefixesRepo = repo;
+    public MonitoredPrefixController(RipeStreamClient client) {
         ripeStreamClient = client;
     }
 
     @PostMapping("/add-one")
     public ResponseEntity<Void> addPrefix(@RequestBody SingleMonitoredPrefixRequest request) {
-        monitoredPrefixesRepo.save(
-                MonitoredPrefix.builder()
-                        .prefix(request.getPrefix())
-                        .createdAt(Instant.now())
-                        .build());
-        // restart the live streamer so we can track this new prefix
-        ripeStreamClient.restartStreamWithPrefixes();
+        // TODO save the prefix to the new kafka topic
+        // We shouldn't have to pass list of prefixes here, find a fix
+        ripeStreamClient.restartStreamWithPrefixes(List.of("8.8.8.0/24"));
         return ResponseEntity.ok().build();
     }
 
     @PostMapping("/add")
     public ResponseEntity<Void> addPrefixes(@RequestBody MultipleMonitoredPrefixesRequest request) {
-        List<MonitoredPrefix> prefixes = request.getPrefixes()
-                .stream()
-                .map(prefix -> MonitoredPrefix.builder().prefix(prefix).createdAt(Instant.now()).build())
-                .toList();
+
         // save in batch
-        monitoredPrefixesRepo.saveAll(prefixes);
+        // TODO save all these prefixes to the kafka topic
+        // TODO do something about restartStream needing the list of prefixes
 
-        ripeStreamClient.restartStreamWithPrefixes();
+        ripeStreamClient.restartStreamWithPrefixes(List.of("8.8.8.0/24"));
 
         return ResponseEntity.ok().build();
     }
 
-    @GetMapping
-    public List<MonitoredPrefix> getPrefixes() {
-        return monitoredPrefixesRepo.findAll();
-    }
+    // TODO Redis implementation for fetching and deleting the prefixes
 
-    @DeleteMapping("/{id}")
-    public ResponseEntity<?> delete(@PathVariable Long id) {
-        monitoredPrefixesRepo.deleteById(id);
-        return ResponseEntity.noContent().build();
-    }
+    // @GetMapping
+    // public List<MonitoredPrefix> getPrefixes() {
+    // return monitoredPrefixesRepo.findAll();
+    // }
+
+    // @DeleteMapping("/{id}")
+    // public ResponseEntity<?> delete(@PathVariable Long id) {
+    // monitoredPrefixesRepo.deleteById(id);
+    // return ResponseEntity.noContent().build();
+    // }
 }

src/main/java/com/nhp/controller/MonitoredPrefixController.java

@@ -1,6 +1,7 @@
 package com.nhp.stream;
 
 import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.List;
@@ -11,37 +12,46 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.nhp.config.StreamProperties;
-import com.nhp.model.MonitoredPrefix;
-import com.nhp.repos.MonitoredPrefixesRepo;
+import com.nhp.services.MetricsService;
 
 import reactor.core.publisher.Mono;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.http.client.WebsocketClientSpec;
 import reactor.core.Disposable;
+import reactor.util.retry.Retry;
 
 @Slf4j
 @Component
 public class RipeStreamClient {
 
     @Autowired
-    private MonitoredPrefixesRepo monitoredPrefixesRepo;
+    private MetricsService metricsService;
 
     private static final String RIS_WS_URL = "wss://ris-live.ripe.net/v1/ws/";
     private final ObjectMapper objectMapper = new ObjectMapper();
     private Disposable connection;
 
     @PostConstruct
     public void startStream() {
-        restartStreamWithPrefixes();
+        // TODO fetch this list of prefixes from the kafka stream
+        restartStreamWithPrefixes(List.of("8.8.8.0/24"));
     }
 
-    public void restartStreamWithPrefixes() {
+    // gracefully handle app shutdown, least thing we can do
+    @PreDestroy
+    public void stopStream() {
+        if (connection != null && !connection.isDisposed()) {
+            connection.dispose();
+            log.info("RIPE stream connection closed on shutdown.");
+        }
+    }
+
+    public void restartStreamWithPrefixes(List<String> prefixes) {
         if (connection != null && !connection.isDisposed()) {
             connection.dispose();
             log.info("Closed previous RIPE stream connection");
         }
-        List<String> prefixes = monitoredPrefixesRepo.findAllPrefixes();
+
         log.info("Connecting to RIPE RIS WebSocket with {} prefix(es)", prefixes.size());
 
         connection = HttpClient.create()
@@ -55,13 +65,19 @@ public void restartStreamWithPrefixes() {
 
                     // consumer logic from this socket now
                     inbound.receive().asString()
-                            .doOnNext(msg -> log.info("BGP Message: {}", msg))
+                            .doOnNext(msg -> {
+                                log.info("BGP Message: {}", msg);
+                                metricsService.incrementBgpMessagesReceieved();
+                            })
                             .doOnError(error -> log.error("Error while streaming", error))
                             .doOnComplete(() -> log.warn("Stream completed/disconnected"))
                             .subscribe();
                     return Mono.never();
                 })
                 .doOnError(error -> log.error("WebSocket connection error", error))
+                .retryWhen(Retry.backoff(5, java.time.Duration.ofSeconds(5))
+                        .doBeforeRetry(retrySignal -> log.warn("Retrying RIPE WebSocket connection (attempt {})",
+                                retrySignal.totalRetries() + 1)))
                 .subscribe();
     }
 

src/main/java/com/nhp/model/MonitoredPrefix.java

@@ -1,26 +1,8 @@
-stream:
-  debounce-ms: 5000
-
-spring:
-  datasource:
-    url: jdbc:postgresql://timescaledb:5432/nhpdb
-    username: nhp
-    password: nhppass
-  jpa:
-    hibernate:
-      ddl-auto: update
-    show-sql: true
-    properties:
-      hibernate:
-        dialect: org.hibernate.dialect.PostgreSQLDialect
-
 logging:
   level:
     root: INFO # Reduce global noise (was DEBUG by default in dev)
     org.springframework: WARN # Suppress most Spring internals
-    org.hibernate.SQL: OFF # Turn off raw SQL logging
-    org.hibernate.type.descriptor.sql.BasicBinder: OFF
-    com.nhp: DEBUG # Your package — keep this at DEBUG to see your own logs
+    com.nhp: INFO # Your package — keep this at DEBUG to see your own logs
 
 management:
   endpoints: