introduce data connectors, used to validate if vin is allowed to connect

.github/workflows/build.yml

@@ -17,16 +17,6 @@ jobs:
       with:
         go-version: 1.20.3
 
-    - name: Set up protoc
-      run: |
-        wget https://github.com/protocolbuffers/protobuf/releases/download/v25.1/protoc-25.1-linux-x86_64.zip
-        unzip protoc-25.1-linux-x86_64.zip
-        sudo mv bin/protoc /usr/local/bin/protoc
-        sudo mv include/* /usr/local/include/
-
-    - name: Install protoc-gen-go
-      run: go install google.golang.org/protobuf/cmd/[email protected]
-
     - name: Generated protofiles are up to date
       run: |
         make generate-protos

Dockerfile.protos

@@ -0,0 +1,44 @@
+FROM debian:12.5-slim
+
+WORKDIR /
+
+ENV GO_VERSION=1.20
+ENV RUBY_VERSION=3.2
+ENV PYTHON_VERSION=3.10
+
+RUN apt-get update && \
+    apt-get install -y \
+    git \
+    wget \
+    unzip \
+    ruby-dev \
+    python3-dev \
+    python3-pip \
+    python3-grpcio \
+    python3-grpc-tools
+
+# protobuf compiler
+ENV PROTOC_VERSION=25.1
+RUN PROTOC_ZIP=protoc-${PROTOC_VERSION}-linux-x86_64.zip && \
+    wget https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/$PROTOC_ZIP -O /tmp/$PROTOC_ZIP && \
+    unzip /tmp/$PROTOC_ZIP -d /usr/local bin/protoc && \
+    unzip /tmp/$PROTOC_ZIP -d /usr/local 'include/*' && \
+    rm -f /tmp/$PROTOC_ZIP
+
+# Go
+ENV PATH="/usr/local/go/bin:$PATH"
+ENV PATH="/root/go/bin:$PATH"
+RUN wget https://dl.google.com/go/go$GO_VERSION.linux-amd64.tar.gz && \
+    tar -C /usr/local -xzf go$GO_VERSION.linux-amd64.tar.gz && \
+    rm go$GO_VERSION.linux-amd64.tar.gz
+RUN go install google.golang.org/protobuf/cmd/[email protected] && \
+    go install google.golang.org/grpc/cmd/[email protected]
+
+# Ruby
+RUN gem install grpc grpc-tools
+
+ENV PROTO_DIR=./protos
+COPY generate_protos.sh /generate_protos.sh
+RUN chmod +x /generate_protos.sh
+
+CMD ["/generate_protos.sh"]

Makefile

@@ -63,19 +63,14 @@ generate-certs:
 clean:
 	find $(PROTO_DIR) -type f ! -name '*.proto' -delete
 
-generate-golang:
-	protoc --go_out=./ --go_opt=paths=source_relative $(PROTO_DIR)/*.proto
+generate-mocks:
+	mockgen -source protos/data_connector_service_grpc.pb.go -destination=test/mocks/data_connector_service_grpc.go -package mocks
 
-generate-python:
-	protoc -I=$(PROTO_DIR) --python_out=$(PROTO_DIR)/python/ $(PROTO_DIR)/*.proto
-
-generate-ruby:
-	protoc --ruby_out=$(PROTO_DIR)/ruby/ --proto_path=$(PROTO_DIR) $(PROTO_FILES)
-
-generate-protos: clean generate-golang generate-python generate-ruby
+generate-protos: clean
+	docker run -v ./protos:/protos:rw --rm -it `docker build -qf Dockerfile.protos .`
 
 image-gen:
 	docker build -t $(ALPHA_IMAGE_NAME) .
 	docker save $(ALPHA_IMAGE_NAME) | gzip > $(ALPHA_IMAGE_COMPRESSED_FILENAME).tar.gz
 
-.PHONY: test build vet linters install integration image-gen generate-protos generate-golang generate-python generate-ruby clean
+.PHONY: test build vet linters install integration image-gen generate-protos generate-mocks clean

README.md

@@ -160,6 +160,21 @@ The following [dispatchers](./telemetry/producer.go#L10-L19) are supported
 
 >NOTE: To add a new dispatcher, please provide integration tests and updated documentation. To serialize dispatcher data as json instead of protobufs, add a config `transmit_decoded_records` and set value to `true` as shown [here](config/test_configs_test.go#L186)
 
+## Data Connectors
+
+Data connectors can be configured to fetch data from a external sources to enhance server functionality.
+
+**Available capabilities**:
+- `vin_allowed`: checks if a vin is allowed to connect to the server. It is recommended to configure this to enhance security.
+
+**Available data connectors**:
+- Redis
+- GRPC
+- HTTP
+- File
+
+To learn about configuring, see their [full documentation](./connector/README.md).
+
 ## Reliable Acks
 Fleet telemetry allows you to send ack messages back to the vehicle. This is useful for applications that need to ensure the data was received and processed. To enable this feature, set `reliable_ack_sources` to one of configured dispatchers (`kafka`,`kinesis`,`pubsub`,`zmq`) in the config file. You can only set reliable acks to one dispatcher per recordType. See [here](./test/integration/config.json#L8) for sample config.
 
@@ -175,9 +190,8 @@ To suppress [tls handshake error logging](https://cs.opensource.google/go/go/+/m
 ## Protos
 Data is encapsulated into protobuf messages of different types. Protos can be recompiled via:
 
-  1. Install protoc, currently on version 4.25.1: https://grpc.io/docs/protoc-installation/
-  2. Install protoc-gen-go: `go install google.golang.org/protobuf/cmd/[email protected]`
-  3. Run make command
+  1. Install Docker
+  2. Run make command
   ```sh
   make generate-protos
   ```

config/config.go

@@ -18,6 +18,7 @@ import (
 	confluent "github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	githublogrus "github.com/sirupsen/logrus"
 
+	"github.com/teslamotors/fleet-telemetry/connector"
 	"github.com/teslamotors/fleet-telemetry/datastore/googlepubsub"
 	"github.com/teslamotors/fleet-telemetry/datastore/kafka"
 	"github.com/teslamotors/fleet-telemetry/datastore/kinesis"
@@ -88,6 +89,12 @@ type Config struct {
 	// TransmitDecodedRecords if true decodes proto message before dispatching it to supported datastores
 	TransmitDecodedRecords bool `json:"transmit_decoded_records,omitempty"`
 
+	// DataConnector defines sources of supplemental data to enhance the server's functionality
+	DataConnectorConfig connector.Config `json:"data_connectors,omitempty"`
+
+	// DataConnector manages accessing supplemental data from external sources
+	DataConnector *connector.ConnectorProvider
+
 	// MetricCollector collects metrics for the application
 	MetricCollector metrics.MetricCollector
 
@@ -235,6 +242,10 @@ func (c *Config) configureMetricsCollector(logger *logrus.Logger) {
 	c.MetricCollector = metrics.NewCollector(c.Monitoring, logger)
 }
 
+func (c *Config) configureDataConnector(logger *logrus.Logger) {
+	c.DataConnector = connector.NewConnectorProvider(c.DataConnectorConfig, c.MetricCollector, logger)
+}
+
 func (c *Config) prometheusEnabled() bool {
 	if c.Monitoring != nil && c.Monitoring.PrometheusMetricsPort > 0 {
 		return true

config/config_initializer.go

@@ -32,6 +32,7 @@ func LoadApplicationConfiguration() (config *Config, logger *logrus.Logger, err
 
 	config.configureLogger(logger)
 	config.configureMetricsCollector(logger)
+	config.configureDataConnector(logger)
 	return config, logger, nil
 }
 

connector/README.md

@@ -0,0 +1,123 @@
+# Data Connectors
+
+Data connectors can be configured to fetch data from an external source to enhance server functionality.
+
+Currently available capabilities:
+- `vin_allowed`: Check if a VIN is allowed to connect to the server.
+
+## Available Connectors
+
+### File
+
+Reads from a JSON file. Watches for file changes at runtime.
+
+- `capabilities`: `[]string` capabilities to use the data connector for.
+- `vin_allowed.path`: `string` path of the file to watch.
+
+**Example File**:
+```json
+{
+    "vins_allowed": ["VIN1"]
+}
+```
+
+**Example Config**:
+```json
+{
+    "data_connectors": {
+        "file": {
+            "capabilities": ["vin_allowed"],
+            "path": "path/to/file"
+        }
+    }
+}
+```
+
+### Redis
+
+Obtains data from a Redis cache.
+
+- `capabilities`: `[]string` capabilities to use the data connector for.
+- `vin_allowed.prefix`: `string` prefix for all Redis keys when checking if a VIN is allowed to connect.
+- `vin_allowed.allow_on_failure`: `bool` whether a VIN should be allowed to connect to the server if an error is encountered while fetching from Redis.
+
+**Example Config**:
+
+```json
+{
+    "data_connectors": {
+        "redis": {
+            "capabilities": ["vin_allowed"],
+            "vin_allowed": {
+                "prefix": "vin_allowed:",
+                "allow_on_failure": true
+            }
+        }
+    }
+}
+```
+
+### HTTP
+
+Obtains data from an REST API.
+
+- `capabilities`: `[]string` capabilities to use the data connector for.
+- `host`: `string` host of the remote server.
+- `timeout_seconds`: `int` seconds to wait for a response.
+- `transport`: `http.Transport` [golang transport options](https://pkg.go.dev/net/http#Transport).
+- `vin_allowed.cache_results`: `bool` whether results from the API should be cached in memory.
+- `vin_allowed.cache_ttl_minutes`: `int` how many minutes each result should be remembered in the cache. Defaults to no expiration.
+- `vin_allowed.allow_on_failure`: `bool` whether a VIN should be allowed to connect to the server if an error is encountered while fetching from Redis.
+
+**Example Config**:
+
+```json
+{
+    "data_connectors": {
+        "http": {
+            "capabilities": ["vin_allowed"],
+            "host": "localhost",
+            "timeout_seconds": 10,
+            "vin_allowed": {
+                "cache_results": true,
+                "cache_ttl_minutes": 60,
+                "allow_on_failure": false
+            }
+        }
+    }
+}
+```
+
+### gRPC
+
+Obtains data from a gRPC service.
+
+- `capabilities`: capabilities to use the data connector for.
+- `host`: host of the gRPC server.
+- `tls.cert_file`: path to cert to use when connecting.
+- `tls.key_file`: path to key to use when connecting.
+- `vin_allowed.cache_results`: `bool` whether results from the API should be cached in memory.
+- `vin_allowed.cache_ttl_minutes`: `int` how many minutes each result should be remembered in the cache. Defaults to no expiration.
+- `vin_allowed.allow_on_failure`: `bool` whether a VIN should be allowed to connect to the server if an error is encountered while fetching from Redis.
+
+**Example Config**:
+
+```json
+{
+    "data_connectors": {
+        "grpc": {
+            "capabilities": ["vin_allowed"],
+            "host": "grpc_host",
+            "tls": {
+                "cert_file": "path/to/cert",
+                "key_file": "path/to/key"
+            },
+            "vin_allowed": {
+                "cache_results": true,
+                "cache_ttl_minutes": 60,
+                "allow_on_failure": false
+            }
+        }
+    }
+}
+```