Connector proxy airbyte

.github/workflows/main.yml

@@ -61,7 +61,6 @@ jobs:
       - run: make install-tools
       - run: go mod download
       - run: make rust-test
-      - run: make go-test-ci
       - run: make rust-binaries
 
       - name: Ensure that generated files are unchanged.
@@ -144,6 +143,19 @@ jobs:
           fetch-depth: 0
           submodules: true
 
+      - name: Install protobuf compiler (it's not already included in CI runner)
+        run: sudo apt install -y libprotobuf-dev protobuf-compiler
+
+      # We require a minimal Go version of 1.17.
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.17.3"
+
+      - name: Install rust toolchain
+        run: rustup show
+      - run: make extra-ci-runner-setup
+      - run: make print-versions
+
       - name: Set up Google Cloud SDK
         uses: google-github-actions/setup-gcloud@v0
         with:
@@ -184,7 +196,20 @@ jobs:
           -o /home/runner/work/flow/flow/.build/package/bin/flowctl \
           -o /home/runner/work/flow/flow/.build/package/bin/flowctl-go \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-connector-proxy \
-          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-proxy \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-tunnel \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-parser \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-schemalate \
+          -o /home/runner/work/flow/flow/.build/package/bin/gazette \
+          -o /home/runner/work/flow/flow/.build/package/bin/sops
+
+      - name: make go-test-ci
+        run: |
+          make go-test-ci \
+          -o /home/runner/work/flow/flow/.build/package/bin/etcd \
+          -o /home/runner/work/flow/flow/.build/package/bin/flowctl \
+          -o /home/runner/work/flow/flow/.build/package/bin/flowctl-go \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-connector-proxy \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-tunnel \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-parser \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-schemalate \
           -o /home/runner/work/flow/flow/.build/package/bin/gazette \
@@ -197,7 +222,7 @@ jobs:
           -o /home/runner/work/flow/flow/.build/package/bin/flowctl \
           -o /home/runner/work/flow/flow/.build/package/bin/flowctl-go \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-connector-proxy \
-          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-proxy \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-tunnel \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-parser \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-schemalate \
           -o /home/runner/work/flow/flow/.build/package/bin/gazette \
@@ -210,7 +235,7 @@ jobs:
           -o /home/runner/work/flow/flow/.build/package/bin/flowctl \
           -o /home/runner/work/flow/flow/.build/package/bin/flowctl-go \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-connector-proxy \
-          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-proxy \
+          -o /home/runner/work/flow/flow/.build/package/bin/flow-network-tunnel \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-parser \
           -o /home/runner/work/flow/flow/.build/package/bin/flow-schemalate \
           -o /home/runner/work/flow/flow/.build/package/bin/gazette \

.vscode/settings.json

@@ -21,6 +21,7 @@
     "files.trimTrailingWhitespace": true,
     "editor.formatOnSave": true,
     "cSpell.words": [
+        "airbyte",
         "Firebolt",
         "schemalate"
     ],

Makefile

@@ -145,9 +145,9 @@ ${RUST_MUSL_BIN}/flow-schemalate:
 ${RUST_MUSL_BIN}/flow-parser:
 	cargo build --target x86_64-unknown-linux-musl --release --locked -p parser
 
-.PHONY: ${RUST_MUSL_BIN}/flow-network-proxy
-${RUST_MUSL_BIN}/flow-network-proxy:
-	cargo build --target x86_64-unknown-linux-musl --release --locked -p network-proxy
+.PHONY: ${RUST_MUSL_BIN}/flow-network-tunnel
+${RUST_MUSL_BIN}/flow-network-tunnel:
+	cargo build --target x86_64-unknown-linux-musl --release --locked -p network-tunnel
 
 .PHONY: ${RUST_MUSL_BIN}/flow-connector-proxy
 ${RUST_MUSL_BIN}/flow-connector-proxy:
@@ -167,7 +167,7 @@ RUST_TARGETS = \
 MUSL_TARGETS = \
 	${PKGDIR}/bin/flow-parser \
 	${PKGDIR}/bin/flow-schemalate \
-	${PKGDIR}/bin/flow-network-proxy \
+	${PKGDIR}/bin/flow-network-tunnel \
 	${PKGDIR}/bin/flow-connector-proxy \
 
 .PHONY: rust-binaries
@@ -193,12 +193,11 @@ ${PKGDIR}/bin/flow-schemalate: ${RUST_MUSL_BIN}/flow-schemalate | ${PKGDIR}
 	cp ${RUST_MUSL_BIN}/flow-schemalate $@
 ${PKGDIR}/bin/flow-parser: ${RUST_MUSL_BIN}/flow-parser | ${PKGDIR}
 	cp ${RUST_MUSL_BIN}/flow-parser $@
-${PKGDIR}/bin/flow-network-proxy: ${RUST_MUSL_BIN}/flow-network-proxy | ${PKGDIR}
-	cp ${RUST_MUSL_BIN}/flow-network-proxy $@
+${PKGDIR}/bin/flow-network-tunnel: ${RUST_MUSL_BIN}/flow-network-tunnel | ${PKGDIR}
+	cp ${RUST_MUSL_BIN}/flow-network-tunnel $@
 ${PKGDIR}/bin/flow-connector-proxy: ${RUST_MUSL_BIN}/flow-connector-proxy | ${PKGDIR}
 	cp ${RUST_MUSL_BIN}/flow-connector-proxy $@
 
-
 ##########################################################################
 # Make targets used by CI:
 
@@ -229,11 +228,11 @@ install-tools: ${PKGDIR}/bin/etcd ${PKGDIR}/bin/sops
 
 .PHONY: rust-test
 rust-test:
-	cargo test --release --locked --workspace --exclude parser --exclude network-proxy --exclude schemalate --exclude connector_proxy
+	cargo test --release --locked --workspace --exclude parser --exclude network-tunnel --exclude schemalate --exclude connector_proxy
 
 .PHONY: musl-test
 musl-test:
-	cargo test --release --locked --target x86_64-unknown-linux-musl --package parser --package network-proxy --package schemalate --package connector_proxy
+	cargo test --release --locked --target x86_64-unknown-linux-musl --package parser --package network-tunnel --package schemalate --package connector_proxy
 
 # `go` test targets must have PATH-based access to tools (etcd & sops),
 # because the `go` tool compiles tests as binaries within a temp directory,
@@ -245,7 +244,7 @@ go-test-fast: $(GO_BUILD_DEPS) | ${PKGDIR}/bin/etcd ${PKGDIR}/bin/sops
 	./go.sh test -p ${NPROC} --tags "${GO_BUILD_TAGS}" ./go/...
 
 .PHONY: go-test-ci
-go-test-ci:   $(GO_BUILD_DEPS) | ${PKGDIR}/bin/etcd ${PKGDIR}/bin/sops
+go-test-ci:   $(GO_BUILD_DEPS) | ${PKGDIR}/bin/etcd ${PKGDIR}/bin/sops ${PKGDIR}/bin/flow-connector-proxy ${PKGDIR}/bin/flowctl ${PKGDIR}/bin/flowctl-go
 	PATH=${PKGDIR}/bin:$$PATH ;\
 	GORACE="halt_on_error=1" ;\
 	./go.sh test -p ${NPROC} --tags "${GO_BUILD_TAGS}" --race --count=15 --failfast ./go/...
@@ -282,4 +281,4 @@ docker-push:
 # It is invoked only for builds on the master branch.
 .PHONY: docker-push-dev
 docker-push-dev:
-	docker push ghcr.io/estuary/flow:dev
+	docker push ghcr.io/estuary/flow:dev

crates/connector_proxy/Cargo.toml

@@ -10,24 +10,29 @@ path = "src/main.rs"
 [dependencies]
 doc = { path = "../doc", version = "0.0.0" }
 flow_cli_common = { path = "../flow_cli_common" }
-network-proxy = { path = "../network-proxy", version = "0.1.0" }
+network-tunnel = { path = "../network-tunnel", version = "0.1.0" }
 protocol = { path = "../protocol", version = "0.0.0" }
 
 async-trait="*"
-async-stream="*"
 bytes = "*"
 byteorder="*"
 clap = { version = "^3", features = ["derive"] }
 futures-core = "*"
 futures-util="*"
+futures="*"
+json-pointer="*"
+libc="*"
 prost = "*"
 schemars = "*"
 serde = { version = "*", features = ["derive"]}
 serde_json = { version = "*", features = ["raw_value"]}
 structopt = "*"
 strum = "*"
 strum_macros = "*"
+tempfile="*"
 thiserror = "*"
 tokio = { version = "1.15.0", features = ["full"] }
 tokio-util = { version = "*", features = ["io"] }
+bytelines = "*"
 tracing="*"
+validator = { version = "*", features = ["derive"] }

crates/connector_proxy/src/apis.rs

@@ -1,9 +1,20 @@
-use crate::errors::Error;
 use bytes::Bytes;
 use clap::ArgEnum;
-use futures_core::stream::Stream;
+use futures::TryStream;
 use std::pin::Pin;
 
+// The protocol used by FlowRuntime to speak with connector-proxy.
+// There are two ways to infer the protocol.
+// 1. From the proxy command passed in from FlowRuntime to the connector proxy.
+// 2. From the connector image labels and tags.
+// The proxy raises an error if both are inconsistent.
+#[derive(Debug, strum_macros::Display, ArgEnum, PartialEq, Clone)]
+#[strum(serialize_all = "snake_case")]
+pub enum FlowRuntimeProtocol {
+    Capture,
+    Materialize,
+}
+
 // Flow Capture operations defined in
 // https://github.com/estuary/flow/blob/master/go/protocols/capture/capture.proto
 #[derive(Debug, strum_macros::Display, ArgEnum, Clone)]
@@ -29,39 +40,12 @@ pub enum FlowMaterializeOperation {
     Transactions,
 }
 
-// To be used as a trait bound for interceptors.
-pub trait FlowOperation {}
-impl FlowOperation for FlowCaptureOperation {}
-impl FlowOperation for FlowMaterializeOperation {}
-
 // An interceptor modifies the request/response streams between Flow runtime and the connector.
 // InterceptorStream defines the type of input and output streams handled by interceptors.
-pub type InterceptorStream = Pin<Box<dyn Stream<Item = std::io::Result<Bytes>> + Send + Sync>>;
-
-// The generic param "T" below is bounded by FlowOperation.
-// A converter is a function that contains the specific stream-handling logic of an interceptor.
-type ConverterFn<T> = Box<dyn Fn(&T, InterceptorStream) -> Result<InterceptorStream, Error>>;
-// An intercept is characterized by a pair of converters, corresponding to the handling logic of request and response streams, respectively.
-pub type RequestResponseConverterPair<T> = (ConverterFn<T>, ConverterFn<T>);
-pub trait Interceptor<T: FlowOperation> {
-    fn get_converters() -> RequestResponseConverterPair<T> {
-        (
-            Box::new(|_op, stream| Ok(stream)),
-            Box::new(|_op, stream| Ok(stream)),
-        )
-    }
-}
-
-// Two converter pairs can be composed together to form a new converter pair.
-pub fn compose<T: 'static + FlowOperation>(
-    a: RequestResponseConverterPair<T>,
-    b: RequestResponseConverterPair<T>,
-) -> RequestResponseConverterPair<T> {
-    let (req_a, resp_a) = a;
-    let (req_b, resp_b) = b;
-    (
-        Box::new(move |op, stream| (req_b)(op, (req_a)(op, stream)?)),
-        // Response conversions are applied in the reverse order of the request conversions.
-        Box::new(move |op, stream| (resp_a)(op, (resp_b)(op, stream)?)),
-    )
-}
+pub type InterceptorStream = Pin<
+    Box<
+        dyn TryStream<Ok = Bytes, Error = std::io::Error, Item = std::io::Result<Bytes>>
+            + Send
+            + Sync,
+    >,
+>;