Operations Guide

1#!/bin/bash
2# backup-streamflow.sh, run on each node
3TIMESTAMP=$(date +%Y%m%d-%H%M%S)
4BACKUP_DIR="/backup/streamflow-$TIMESTAMP"
5DATA_DIR="./data"
6
7# 1. Graceful stop (flush deferred writes)
8pkill -SIGTERM -f StreamFlowApplication
9sleep 10
10
11# 2. Copy data
12mkdir -p "$BACKUP_DIR"
13cp -r "$DATA_DIR" "$BACKUP_DIR/data"
14cp streamflow-app/src/main/resources/streamflow.properties "$BACKUP_DIR/"
15
16# 3. Compress
17tar -czf "/backup/streamflow-$TIMESTAMP.tar.gz" -C /backup "streamflow-$TIMESTAMP"
18rm -rf "$BACKUP_DIR"
19
20echo "Backup: /backup/streamflow-$TIMESTAMP.tar.gz ($(du -h /backup/streamflow-$TIMESTAMP.tar.gz | cut -f1))"
21
22# 4. Restart
23./scripts/start-streamflow.sh

1#!/bin/bash
2# backup-incremental.sh, no need to stop the server
3TIMESTAMP=$(date +%Y%m%d-%H%M%S)
4BACKUP_DIR="/backup/incremental-$TIMESTAMP"
5DATA_DIR="./data"
6
7# Closed segments (immutable) can be copied while running
8# Only the active segment is being written to
9mkdir -p "$BACKUP_DIR"
10
11for partition_dir in "$DATA_DIR"/*/; do
12    topic_partition=$(basename "$partition_dir")
13    mkdir -p "$BACKUP_DIR/$topic_partition"
14
15    # Copy all segments EXCEPT the last (active) one
16    # Closed segments are immutable → safe to copy
17    segments=($(ls "$partition_dir"/*.log 2>/dev/null | sort))
18    count=${#segments[@]}
19
20    if [ "$count" -gt 1 ]; then
21        for ((i=0; i<count-1; i++)); do
22            base=$(basename "${segments[$i]}" .log)
23            cp "$partition_dir/$base".{log,index,timeindex} "$BACKUP_DIR/$topic_partition/" 2>/dev/null
24        done
25    fi
26done
27
28echo "Incremental backup: $BACKUP_DIR"

1#!/bin/bash
2# restore-streamflow.sh
3BACKUP="/backup/streamflow-20260401-120000.tar.gz"
4
5# 1. Stop StreamFlow
6pkill -f StreamFlowApplication
7sleep 5
8
9# 2. Extract backup
10tar -xzf "$BACKUP" -C /tmp/
11
12# 3. Replace data
13rm -rf ./data
14cp -r /tmp/streamflow-*/data ./data
15cp /tmp/streamflow-*/streamflow.properties streamflow-app/src/main/resources/
16
17# 4. Restart (Raft recovery rebuilds state)
18./scripts/start-streamflow.sh
19
20echo "Restore complete. Verify: curl http://localhost:8080/health"

1#!/bin/bash
2# rolling-upgrade.sh, run from a control station
3NODES=("streamflow-1" "streamflow-2" "streamflow-3")
4NEW_VERSION="0.2.0"
5
6for node in "${NODES[@]}"; do
7    echo "═══ Upgrading $node ═══"
8
9    # 1. Check cluster health before
10    echo "  Cluster health check..."
11    ssh "$node" "curl -sf http://localhost:8080/health || exit 1"
12
13    # 2. Drain the node (clients switch to others)
14    echo "  Stopping $node..."
15    ssh "$node" "pkill -SIGTERM -f StreamFlowApplication"
16    sleep 15  # allow time for Raft re-election if it was the leader
17
18    # 3. Update the code
19    echo "  Pulling new version..."
20    ssh "$node" "cd ~/streamflow && git pull origin dev"
21    ssh "$node" "cd ~/streamflow && mvn package -DskipTests -q"
22
23    # 4. Restart
24    echo "  Starting $node with v$NEW_VERSION..."
25    ssh "$node" "cd ~/streamflow && ./scripts/start-streamflow.sh"
26
27    # 5. Wait for the node to rejoin the cluster
28    echo "  Waiting for $node to rejoin..."
29    for i in $(seq 1 60); do
30        if ssh "$node" "curl -sf http://localhost:8080/ready" > /dev/null 2>&1; then
31            echo "  ✓ $node is ready (${i}s)"
32            break
33        fi
34        sleep 1
35    done
36
37    # 6. Check replication
38    echo "  Checking replication..."
39    sleep 10
40    ssh "$node" "curl -s http://localhost:8080/metrics | grep replication_lag"
41
42    echo ""
43done
44
45echo "Rolling upgrade complete. All nodes on v$NEW_VERSION"

1# If a node fails to start after upgrade
2ssh streamflow-1 "cd ~/streamflow && git checkout v1.0.0"
3ssh streamflow-1 "cd ~/streamflow && mvn package -DskipTests -q"
4ssh streamflow-1 "cd ~/streamflow && ./scripts/start-streamflow.sh"

1#!/bin/bash
2# daily-check.sh
3echo "═══ StreamFlow Daily Health Check ═══"
4echo ""
5
6# 1. Are all nodes UP?
7for node in streamflow-{1,2,3}; do
8    status=$(curl -sf http://$node:8080/health | jq -r '.status' 2>/dev/null || echo "DOWN")
9    echo "  $node: $status"
10done
11
12# 2. Raft replication OK?
13echo ""
14echo "  Replication lag:"
15curl -s http://streamflow-1:8080/metrics | grep replication_lag | head -3
16
17# 3. Consumer lag OK?
18echo ""
19echo "  Consumer lag:"
20curl -s http://streamflow-1:8080/metrics | grep consumer_lag | head -5
21
22# 4. Disk space
23echo ""
24echo "  Disk usage:"
25for node in streamflow-{1,2,3}; do
26    usage=$(ssh $node "df -h /opt/streamflow/data | tail -1 | awk '{print \$5}'" 2>/dev/null || echo "N/A")
27    echo "    $node: $usage"
28done
29
30# 5. Recent errors
31echo ""
32echo "  Recent errors (last 1h):"
33grep -c "ERROR" /tmp/streamflow.log 2>/dev/null || echo "    0"

1# logrotate config: /etc/logrotate.d/streamflow
2/tmp/streamflow.log {
3    daily
4    rotate 7
5    compress
6    delaycompress
7    missingok
8    notifempty
9    copytruncate
10}

1# Compaction is automatic (configurable in streamflow.properties)
2# To force a manual compaction:
3curl -X POST http://localhost:8080/api/v1/admin/compact?topic=orders

1# 1. Provision a new VM
2# 2. Install StreamFlow
3# 3. Configure with the same bootstrap.servers
4# 4. Start → the node synchronizes automatically via Raft

1# 1. Restart one of the 2 downed nodes
2# 2. Wait for Raft re-election (~15s)
3# 3. Cluster resumes with 2/3 nodes
4# 4. Replace the 3rd node when possible

1# 1. Provision 3 new VMs
2# 2. Restore the most recent backup on each node
3# 3. Start the cluster
4# 4. Data since the last backup is lost
5#    → If geo-replication enabled: restore from secondary region

1# 1. Stop the corrupted node
2# 2. Delete its data: rm -rf data/
3# 3. Restart → it resynchronizes from other nodes via Raft

1# 1. Identify live nodes
2for node in streamflow-{1,2,3}; do
3    nc -z $node 9092 2>/dev/null && echo "$node: UP" || echo "$node: DOWN"
4done
5
6# 2. If no node UP → restart the most recent one
7ssh streamflow-1 "./scripts/start-streamflow.sh"
8# Wait 30s
9ssh streamflow-2 "./scripts/start-streamflow.sh"
10ssh streamflow-3 "./scripts/start-streamflow.sh"
11
12# 3. Verify
13curl http://streamflow-1:8080/health

1# 1. Identify the source
2curl -s http://localhost:8080/metrics | grep -E "fsync|publish_latency|gc_pause"
3
4# If fsync_latency > 100ms → disk issue
5    # Check IOPS: iostat -x 1 5
6    # Solution: switch to local NVMe
7
8# If gc_pause > 1s → memory issue
9    # Check heap: curl -s http://localhost:8080/metrics | grep heap
10    # Solution: increase -Xmx
11
12# If publish_latency OK but consumer_lag growing → consumers too slow
13    # Solution: scale the consumers

1# 1. Check space
2df -h /opt/streamflow/data
3
4# 2. Identify large topics
5du -sh data/*/ | sort -rh | head -10
6
7# 3. Options:
8#    a. Reduce retention
9#    b. Compact topics
10#    c. Add disk
11#    d. Delete test topics
12curl -X DELETE http://localhost:8080/api/v1/admin/topics/test-bench-*

1# 1. Check if process is dead
2pgrep -f StreamFlowApplication || echo "PROCESS DEAD"
3
4# 2. Analyze heap dump (if -XX:+HeapDumpOnOutOfMemoryError)
5ls -la /tmp/*.hprof
6
7# 3. Increase memory and restart
8export JAVA_OPTS="-Xms4g -Xmx12g"  # was -Xmx8g
9./scripts/start-streamflow.sh
10
11# 4. Investigate root cause
12#    - FetchBuffer too large? (64MB × partitions)
13#    - Too many agents? (each consumes ~10MB)
14#    - Memory leak? (heap dump analysis with Eclipse MAT)

1# 1. Provision the VM
2# 2. Install StreamFlow
3# 3. Configure (add the new node to bootstrap)
4cat >> streamflow.properties << 'EOF'
5bootstrap.servers=node-1:sf-1:5679,node-2:sf-2:5680,node-3:sf-3:5681,node-4:sf-4:5682
6EOF
7
8# 4. Start
9./scripts/start-streamflow.sh
10
11# 5. Verify the join
12curl http://sf-4:8080/health
13# Slots will be automatically rebalanced (~25% migrate to the new node)