Add support for feeder_code, add pw-ingest to feed-in (#36)

* add pw_ingest to mux

* add pw_ingest to mux

* bump debian ver

* start pw-ingest on start

* dont start pw-ingest yet

* start pw-ingest for mlat results

* show version of pw_ingest

* add adsc-receiver

* add ports to adsc-receiver

* --ads-c

* put adsc-receiver in feeder net

* work on new feeder id

* fix struct naming

* work on feeder_code

* remove unused code

* add feeder_code

* fix template panic

* bump versions

* update tests

* linting

* feeder_code -> FeederCode

* fix healthcheck

* reduce healthcheck frequency

* random sleep for healthcheck

* limit container start requests

* limit container start requests

* start the container with timeout

* remove comment

* remove pointless code

* remove pointless code

* network changes

* network changes

* network changes

* undo network changes

* remove readsb from feed-in & mux containers

* configuration via variables

* update env vars for feed-in containers

* remove readsb args

* comments

* update healthcheck & env defaults

* fix healthcheck

* remove adsc

* remove adsc, add PW_INGEST_ENABLE_ADSC

* fix test

* fix test

* fix test

Dockerfile.bordercontrol

@@ -1,4 +1,4 @@
-FROM golang:1.21.4-bookworm AS builder
+FROM golang:1.21.5-bookworm AS builder
 
 COPY / /src/pw-bordercontrol
 
@@ -8,7 +8,7 @@ RUN go mod tidy && \
     go build ./... && \
     go install ./...
 
-FROM debian:bookworm-20231030-slim
+FROM debian:bookworm-20231120-slim
 
 ARG S6_OVERLAY_VERSION=3.1.6.0
 

Dockerfile.feeder

@@ -1,6 +1,6 @@
 FROM ghcr.io/plane-watch/pw-pipeline:pw_ingest as pipeline
 
-FROM debian:bookworm-20231030-slim
+FROM debian:bookworm-20231120-slim
 
 ARG S6_OVERLAY_VERSION=3.1.6.0
 
@@ -16,42 +16,11 @@ RUN set -x && \
     KEPT_PACKAGES+=(procps) && \
     KEPT_PACKAGES+=(iproute2) && \
     KEPT_PACKAGES+=(xz-utils) && \
-    # packages needed to build
-    TEMP_PACKAGES+=(build-essential) && \
-    TEMP_PACKAGES+=(cmake) && \
-    TEMP_PACKAGES+=(pkg-config) && \
-    # prerequisites for readsb-protobuf \
-    KEPT_PACKAGES+=(protobuf-c-compiler) && \
-    KEPT_PACKAGES+=(libprotobuf-c1) && \
-    TEMP_PACKAGES+=(libprotobuf-c-dev) && \
-    TEMP_PACKAGES+=(librrd-dev) && \
-    KEPT_PACKAGES+=(libncurses6) && \
-    TEMP_PACKAGES+=(libncurses5-dev) && \
-    TEMP_PACKAGES+=(binutils) && \
     apt-get update && \
     apt-get install -y --no-install-recommends \
         "${KEPT_PACKAGES[@]}" \
         "${TEMP_PACKAGES[@]}" \
         && \
-    # install mictronics readsb
-    BRANCH_READSB="dev" && \
-    # readsb: clone repo
-    git clone \
-      --branch "$BRANCH_READSB" \
-      --depth 1 \
-      --single-branch \
-      'https://github.com/Mictronics/readsb-protobuf.git' \
-      /src/readsb-protobuf \
-      && \
-    # readsb: build & install (note, -j seems to have issues, so not using...)
-    pushd /src/readsb-protobuf && \
-    make BLADERF=no RTLSDR=no PLUTOSDR=no HAVE_BIASTEE=no && \
-    find "/src/readsb-protobuf" -maxdepth 1 -executable -type f -exec cp -v {} /usr/local/bin/ \; && \
-    popd && \
-    ldconfig && \
-    # readsb: simple tests
-    readsb --version && \
-    viewadsb --version && \
     # deploy healthchecks framework
     git clone \
       --depth=1 \
@@ -81,7 +50,13 @@ RUN tar -C / -Jxpf /tmp/s6-overlay-x86_64.tar.xz
 ENV S6_LOGGING=0 \
     S6_VERBOSITY=1
 
+# Set defaults
+ENV PW_INGEST_INPUT_MODE=listen \
+    PW_INGEST_INPUT_PROTO=beast \
+    PW_INGEST_INPUT_ADDR=0.0.0.0 \
+    PW_INGEST_INPUT_PORT=12345
+
 # Set s6 init as entrypoint
 ENTRYPOINT [ "/init" ]
 
-HEALTHCHECK --interval=60s --timeout=10s --start-period=300s --retries=5 CMD /scripts/healthcheck.sh
+HEALTHCHECK --interval=300s --timeout=40s --start-period=600s --retries=5 CMD /scripts/healthcheck.sh

Dockerfile.mux

@@ -1,4 +1,6 @@
-FROM debian:bookworm-20231030-slim
+FROM ghcr.io/plane-watch/pw-pipeline:pw_ingest as pipeline
+
+FROM debian:bookworm-20231120-slim
 
 ARG S6_OVERLAY_VERSION=3.1.6.0
 
@@ -87,6 +89,7 @@ RUN set -x && \
     apt-get autoremove -y && \
     rm -rf /src/* /tmp/* /var/lib/apt/lists/*
 
+COPY --from=pipeline /app/pw_ingest /usr/local/bin/pw_ingest
 COPY pw-mux/rootfs/ /
 
 # Deploy s6 overlay

docker-compose.yml

@@ -15,12 +15,7 @@ x-mux-common: &mux-common
   networks:
     - bordercontrol_feeder
   environment:
-    - READSB_NET_BEAST_OUTPUT_PORT=30005
-    - READSB_NET_BEAST_REDUCE_OUT_PORT=30015
-    - READSB_NET_BEAST_REDUCE_INTERVAL=0.5
-    - READSB_NET_BEAST_INPUT_PORT=12345
-    - READSB_NET_ENABLE=true
-    - READSB_NET_ONLY=true
+    - PW_INGEST_SINK=${PW_INGEST_SINK}
   tmpfs:
     - /run:exec,size=64M
     - /var/log
@@ -55,6 +50,7 @@ services:
       - 12345:12345    # BEAST
       - 12346:12346    # MLAT
 
+  # feed-in image builder
   feed-in-builder:
     container_name: feed-in-builder
     build:
@@ -63,98 +59,74 @@ services:
     image: feed-in
     entrypoint: /bin/true
 
-  # Multiplexer: Australia ACT
+  # MLAT: Australia ACT
   mux-act:
     container_name: mux-act
     hostname: mux-act
     <<: *mux-common
-    ports:
-      - 12601:30015
 
-  # Multiplexer: Australia NSW
+  # MLAT: Australia NSW
   mux-nsw:
     container_name: mux-nsw
     hostname: mux-nsw
     <<: *mux-common
-    ports:
-      - 12001:30015
 
-  # Multiplexer: Australia NT
+  # MLAT: Australia NT
   mux-nt:
     container_name: mux-nt
     hostname: mux-nt
     <<: *mux-common
-    ports:
-      - 10801:30015
 
-  # Multiplexer: QLD
+  # MLAT: QLD
   mux-qld:
     container_name: mux-qld
     hostname: mux-qld
     <<: *mux-common
-    ports:
-      - 14001:30015
 
-  # Multiplexer: SA
+  # MLAT: SA
   mux-sa:
     container_name: mux-sa
     hostname: mux-sa
     <<: *mux-common
-    ports:
-      - 15001:30015
 
-  # Multiplexer: TAS
+  # MLAT: TAS
   mux-tas:
     container_name: mux-tas
     hostname: mux-tas
     <<: *mux-common
-    ports:
-      - 17001:30015
 
-  # Multiplexer: VIC
+  # MLAT: VIC
   mux-vic:
     container_name: mux-vic
     hostname: mux-vic
     <<: *mux-common
-    ports:
-      - 13001:30015
 
-  # Multiplexer: WA
+  # MLAT: WA
   mux-wa:
     container_name: mux-wa
     hostname: mux-wa
     <<: *mux-common
-    ports:
-      - 16001:30015
 
-  # Multiplexer: NZ
+  # MLAT: NZ
   mux-nz:
     container_name: mux-nz
     hostname: mux-nz
     <<: *mux-common
-    ports:
-      - 19001:30015
 
-  # Multiplexer: EU
+  # MLAT: EU
   mux-eu:
     container_name: mux-eu
     hostname: mux-eu
     <<: *mux-common
-    ports:
-      - 22001:30015
 
-  # Multiplexer: USA
+  # MLAT: USA
   mux-us:
     container_name: mux-us
     hostname: mux-us
     <<: *mux-common
-    ports:
-      - 23001:30015
 
-  # Multiplexer: ASIA
+  # MLAT: ASIA
   mux-asia:
     container_name: mux-asia
     hostname: mux-asia
     <<: *mux-common
-    ports:
-      - 24001:30015

pw-bordercontrol/cmd/bordercontrol/containers.go

@@ -184,6 +184,7 @@ func startFeederContainers(
 			Str("label", string(containerToStart.clientDetails.label)).
 			Str("uuid", string(containerToStart.clientDetails.clientApiKey.String())).
 			Str("src", string(containerToStart.srcIP.String())).
+			Str("code", string(containerToStart.clientDetails.feederCode)).
 			Logger()
 
 		// determine if container is already running
@@ -222,13 +223,12 @@ func startFeederContainers(
 				fmt.Sprintf("FEEDER_LAT=%f", containerToStart.clientDetails.refLat),
 				fmt.Sprintf("FEEDER_LON=%f", containerToStart.clientDetails.refLon),
 				fmt.Sprintf("FEEDER_UUID=%s", containerToStart.clientDetails.clientApiKey.String()),
-				// "READSB_STATS_EVERY=300",
-				"READSB_NET_ENABLE=true",
-				"READSB_NET_BEAST_INPUT_PORT=12345",
-				"READSB_NET_BEAST_OUTPUT_PORT=30005",
-				"READSB_NET_ONLY=true",
-				fmt.Sprintf("READSB_NET_CONNECTOR=%s,12345,beast_out", containerToStart.clientDetails.mux),
+				fmt.Sprintf("FEEDER_TAG=%s", containerToStart.clientDetails.feederCode),
 				"PW_INGEST_PUBLISH=location-updates",
+				"PW_INGEST_INPUT_MODE=listen",
+				"PW_INGEST_INPUT_PROTO=beast",
+				"PW_INGEST_INPUT_ADDR=0.0.0.0",
+				"PW_INGEST_INPUT_PORT=12345",
 				fmt.Sprintf("PW_INGEST_SINK=%s", pwIngestPublish),
 			}
 

pw-bordercontrol/cmd/bordercontrol/containers_test.go

@@ -32,19 +32,14 @@ const (
 	TestFeederLatitude  = 123.456789
 	TestFeederLongitude = 98.765432
 	TestFeederMux       = "test-mux"
+	TestFeederCode      = "ABCD-1234"
 
 	TestPWIngestSink = "nats://pw-ingest-sink:12345"
 )
 
 func TestContainersWithKill(t *testing.T) {
 
 	var (
-		ContainerEnvVarFeederLatOK                bool
-		ContainerEnvVarFeederLonOK                bool
-		ContainerEnvVarFeederUUIDOK               bool
-		ContainerEnvVarFeederReadsbNetConnectorOK bool
-		ContainerEnvVarFeederPWIngestSinkOK       bool
-
 		ContainerNetworkOK bool
 	)
 
@@ -124,7 +119,8 @@ func TestContainersWithKill(t *testing.T) {
 			refLat:       TestFeederLatitude,
 			refLon:       TestFeederLongitude,
 			mux:          TestFeederMux,
-			label:        TestFeederMux,
+			label:        TestFeederLabel,
+			feederCode:   TestFeederCode,
 		},
 		srcIP: net.IPv4(127, 0, 0, 1),
 	}
@@ -143,25 +139,21 @@ func TestContainersWithKill(t *testing.T) {
 
 	// check environment variables
 	t.Log("checking container environment variables")
+	envVars := make(map[string]string)
 	for _, e := range ct.Config.Env {
-		switch e {
-		case fmt.Sprintf("FEEDER_LAT=%f", TestFeederLatitude):
-			ContainerEnvVarFeederLatOK = true
-		case fmt.Sprintf("FEEDER_LON=%f", TestFeederLongitude):
-			ContainerEnvVarFeederLonOK = true
-		case fmt.Sprintf("FEEDER_UUID=%s", strings.ToLower(TestFeederAPIKey)):
-			ContainerEnvVarFeederUUIDOK = true
-		case fmt.Sprintf("READSB_NET_CONNECTOR=%s,12345,beast_out", TestFeederMux):
-			ContainerEnvVarFeederReadsbNetConnectorOK = true
-		case fmt.Sprintf("PW_INGEST_SINK=%s", TestPWIngestSink):
-			ContainerEnvVarFeederPWIngestSinkOK = true
-		}
+		envVars[strings.Split(e, "=")[0]] = strings.Split(e, "=")[1]
 	}
-	assert.True(t, ContainerEnvVarFeederLatOK)
-	assert.True(t, ContainerEnvVarFeederLonOK)
-	assert.True(t, ContainerEnvVarFeederUUIDOK)
-	assert.True(t, ContainerEnvVarFeederReadsbNetConnectorOK)
-	assert.True(t, ContainerEnvVarFeederPWIngestSinkOK)
+
+	assert.Equal(t, fmt.Sprintf("%f", TestFeederLatitude), envVars["FEEDER_LAT"])
+	assert.Equal(t, fmt.Sprintf("%f", TestFeederLongitude), envVars["FEEDER_LON"])
+	assert.Equal(t, strings.ToLower(fmt.Sprintf("%s", TestFeederAPIKey)), envVars["FEEDER_UUID"])
+	assert.Equal(t, TestFeederCode, envVars["FEEDER_TAG"])
+	assert.Equal(t, TestPWIngestSink, envVars["PW_INGEST_SINK"])
+	assert.Equal(t, "location-updates", envVars["PW_INGEST_PUBLISH"])
+	assert.Equal(t, "listen", envVars["PW_INGEST_INPUT_MODE"])
+	assert.Equal(t, "beast", envVars["PW_INGEST_INPUT_PROTO"])
+	assert.Equal(t, "0.0.0.0", envVars["PW_INGEST_INPUT_ADDR"])
+	assert.Equal(t, "12345", envVars["PW_INGEST_INPUT_PORT"])
 
 	// check container autoremove set to true
 	t.Log("check container autoremove set to true")

pw-bordercontrol/cmd/bordercontrol/feeder_conn.go

@@ -32,9 +32,9 @@ type (
 
 	// struct to hold feeder client information
 	feederClient struct {
-		clientApiKey   uuid.UUID
-		refLat, refLon float64
-		mux, label     string
+		clientApiKey           uuid.UUID
+		refLat, refLon         float64
+		mux, label, feederCode string
 	}
 
 	// struct for proxy goroutines
@@ -298,7 +298,10 @@ func authenticateFeeder(connIn net.Conn) (clientDetails *feederClient, err error
 	if err != nil {
 		return clientDetails, err
 	}
-	log = log.With().Str("uuid", clientDetails.clientApiKey.String()).Logger()
+	log = log.With().
+		Str("uuid", clientDetails.clientApiKey.String()).
+		Str("code", clientDetails.feederCode).
+		Logger()
 	// log.Trace().Msg("feeder API key received from SNI")
 
 	// check valid api key against atc
@@ -500,6 +503,7 @@ func proxyClientConnection(connIn net.Conn, connProto string, connNum uint, cont
 		Str("uuid", clientDetails.clientApiKey.String()).
 		Str("mux", clientDetails.mux).
 		Str("label", clientDetails.label).
+		Str("code", clientDetails.feederCode).
 		Logger()
 
 	// check number of connections, and drop connection if limit exceeded
@@ -519,12 +523,20 @@ func proxyClientConnection(connIn net.Conn, connProto string, connNum uint, cont
 	switch connProto {
 	case protoBeast:
 
-		log = log.With().Str("dst", fmt.Sprintf("%s%s", feedInContainerPrefix, clientDetails.clientApiKey.String())).Logger()
+		log = log.With().
+			Str("dst", fmt.Sprintf("%s%s", feedInContainerPrefix, clientDetails.clientApiKey.String())).
+			Logger()
 
-		// start the container
-		containersToStartRequests <- startContainerRequest{
+		// request start of the feed-in container with submission timeout
+		select {
+		case containersToStartRequests <- startContainerRequest{
 			clientDetails: clientDetails,
 			srcIP:         remoteIP,
+		}:
+		case <-time.After(5 * time.Second):
+			err := errors.New("5s timeout waiting to submit container start request")
+			log.Err(err).Msg("could not start feed-in container")
+			return err
 		}
 
 		// wait for request to be actioned
@@ -597,7 +609,7 @@ func proxyClientConnection(connIn net.Conn, connProto string, connNum uint, cont
 	}
 
 	// update stats
-	stats.addConnection(clientDetails.clientApiKey, connIn.RemoteAddr(), connOut.RemoteAddr(), connProto, connNum)
+	stats.addConnection(clientDetails.clientApiKey, connIn.RemoteAddr(), connOut.RemoteAddr(), connProto, clientDetails.feederCode, connNum)
 	defer stats.delConnection(clientDetails.clientApiKey, connProto, connNum)
 
 	// method to signal goroutines to exit