feat(sgx): add sidecar dockerfile

Author: Gezq

sgx_network_simulation/Dockerfile

@@ -0,0 +1,33 @@
+FROM golang:1.16 AS go
+
+RUN apt-get update && \
+    apt-get install -y make g++ libgmp-dev libglib2.0-dev libssl-dev && \
+    apt-get install -y protobuf-compiler && \
+    apt-get clean
+
+WORKDIR /app
+COPY tools/tcp_grpc_proxy ./
+RUN make build
+
+FROM python:3.6.8
+
+RUN echo "deb http://archive.debian.org/debian stretch main contrib non-free" > /etc/apt/sources.list
+
+RUN apt-get update && \
+    apt-get install -y curl vim make nginx && \
+    apt-get clean
+
+# upgrade nginx
+RUN echo "deb http://nginx.org/packages/mainline/debian/ stretch nginx deb-src http://nginx.org/packages/mainline/debian/ stretch nginx" > /etc/apt/sources.list.d/nginx.list
+RUN wget -qO - https://nginx.org/keys/nginx_signing.key | apt-key add -
+RUN apt update && \
+    apt remove nginx-common -y && \
+    apt install nginx
+
+COPY sgx_network_simulation/ /app/
+WORKDIR /app
+COPY --from=go /app/tcp2grpc ./
+COPY --from=go /app/grpc2tcp ./
+RUN pip3 install -r requirements.txt && make protobuf
+
+ENTRYPOINT ["bash", "docker_entrypoint.sh"]

sgx_network_simulation/nginx/sidecar.conf

@@ -0,0 +1,22 @@
+# Forwards all traffic to nginx controller
+server {
+    listen 32102 http2;
+
+    # No limits
+    client_max_body_size 0;
+    grpc_read_timeout 3600s;
+    grpc_send_timeout 3600s;
+    client_body_timeout 3600s;
+    # grpc_socket_keepalive is recommended but not required
+    # grpc_socket_keepalive is supported after nginx 1.15.6
+    grpc_socket_keepalive on;
+
+    grpc_set_header Authority fl-bytedance-client-auth.com;
+    grpc_set_header Host fl-bytedance-client-auth.com;
+    grpc_set_header X-Host sgx-test.fl-cmcc.com;
+
+    location / {
+        # Redirects to nginx controller
+        grpc_pass grpc://fedlearner-stack-ingress-nginx-controller.default.svc:80;
+    }
+}

sgx_network_simulation/sidecar.sh

@@ -0,0 +1,66 @@
+#!/bin/bash
+set -ex
+
+FILE_PATH="/pod-data/listen_port"
+while [ ! -s "$FILE_PATH" ]; do
+  echo "wait for $FILE_PATH ..."
+  sleep 1
+done
+WORKER_LISTEN_PORT=$(cat "$FILE_PATH")
+
+echo "# Forwards all traffic to nginx controller
+server {
+    listen 32102 http2;
+
+    # No limits
+    client_max_body_size 0;
+    grpc_read_timeout 3600s;
+    grpc_send_timeout 3600s;
+    client_body_timeout 3600s;
+    # grpc_socket_keepalive is recommended but not required
+    # grpc_socket_keepalive is supported after nginx 1.15.6
+    grpc_socket_keepalive on;
+
+    grpc_set_header Authority ${EGRESS_HOST};
+    grpc_set_header Host ${EGRESS_HOST};
+    grpc_set_header X-Host ${SERVICE_ID}.${EGRESS_DOMAIN};
+
+    location / {
+        # Redirects to nginx controller
+        grpc_pass grpc://fedlearner-stack-ingress-nginx-controller.default.svc:80;
+    }
+}
+" > nginx/sidecar.conf
+
+if [ -z "$PORT0" ]; then
+  PORT0=32001
+fi
+
+if [ -z "$PORT2" ]; then
+  PORT2=32102
+fi
+
+sed -i "s/listen [0-9]* http2;/listen $PORT2 http2;/" nginx/sidecar.conf
+
+cp nginx/sidecar.conf /etc/nginx/conf.d/
+service nginx restart
+
+# Server sidecar: grpc to tcp, 5001 is the server port of main container
+echo "Starting server sidecar"
+./grpc2tcp --grpc_server_port=$PORT0 \
+           --target_tcp_address="localhost:$WORKER_LISTEN_PORT" &
+
+echo "Starting client sidecar"
+./tcp2grpc --tcp_server_port="$PROXY_LOCAL_PORT" \
+           --target_grpc_address="localhost:$PORT2" &
+
+echo "===========Sidecar started!!============="
+
+while true
+do
+  if [[ -f "/pod-data/main-terminated" ]]
+  then
+    exit 0
+  fi
+  sleep 5
+done

tools/tcp_grpc_proxy/Makefile

@@ -0,0 +1,13 @@
+install:
+	go get tcp_grpc_proxy
+	go mod download
+
+protobuf: install
+	go install google.golang.org/protobuf/cmd/[email protected]
+	go install google.golang.org/grpc/cmd/[email protected]
+	PATH="${PATH}:$(shell go env GOPATH)/bin" \
+		protoc -I=proto --go_out=. --go-grpc_out=. proto/*.proto
+
+build: protobuf
+	go build -o tcp2grpc cmd/tcp2grpc/main.go
+	go build -o grpc2tcp cmd/grpc2tcp/main.go

tools/tcp_grpc_proxy/cmd/grpc2tcp/main.go

@@ -1,9 +1,9 @@
 package main
 
 import (
-	"fedlearner.net/tools/tcp_grpc_proxy/pkg/proxy"
 	"flag"
 	"fmt"
+	"tcp_grpc_proxy/proxy"
 )
 
 func main() {
@@ -14,6 +14,6 @@ func main() {
 	flag.Parse()
 	grpcServerAddress := fmt.Sprintf("0.0.0.0:%d", grpcServerPort)
 
-	grpc2tcpServer := proxy.NewGrpc2TcpServer(grpcServerAddress, targetTCPAddress)
+	grpc2tcpServer := proxy.NewGrpc2TCPServer(grpcServerAddress, targetTCPAddress)
 	grpc2tcpServer.Run()
 }

tools/tcp_grpc_proxy/cmd/grpcclient/main.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"time"
+
+	"tcp_grpc_proxy/proto"
+
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+)
+
+func main() {
+	// Set up a connection to the server.
+	grpcServer := "127.0.0.1:7766"
+	conn, err := grpc.Dial(grpcServer, grpc.WithInsecure())
+	if err != nil {
+		logrus.Fatalf("did not connect: %v", err)
+	}
+	defer conn.Close()
+	tsc := proto.NewTunnelServiceClient(conn)
+
+	tc, err := tsc.Tunnel(context.Background())
+	if err != nil {
+		logrus.Fatalln(err)
+	}
+
+	sendPacket := func(data []byte) error {
+		return tc.Send(&proto.Chunk{Data: data})
+	}
+
+	go func() {
+		for {
+			chunk, err := tc.Recv()
+			if err != nil {
+				logrus.Println("Recv terminated:", err)
+				os.Exit(0)
+			}
+			logrus.Println(string(chunk.Data))
+		}
+
+	}()
+
+	for {
+		time.Sleep(time.Duration(2) * time.Second)
+		buf := bytes.NewBufferString("************Hello World**********").Bytes()
+		sendPacket(buf)
+	}
+}

tools/tcp_grpc_proxy/cmd/grpcserver/main.go

@@ -0,0 +1,11 @@
+package main
+
+import (
+	"tcp_grpc_proxy/grpc2tcp"
+)
+
+func main() {
+	grpcServerAddress := "0.0.0.0:7766"
+	targetTCPAddress := "127.0.0.1:17766"
+	grpc2tcp.RunServer(grpcServerAddress, targetTCPAddress)
+}

tools/tcp_grpc_proxy/cmd/tcp2grpc/main.go

@@ -1,11 +1,49 @@
 package main
 
 import (
-	"fedlearner.net/tools/tcp_grpc_proxy/pkg/proxy"
 	"flag"
 	"fmt"
+	"io"
+	"net"
+	"os"
+	"tcp_grpc_proxy/proxy"
 )
 
+func test() {
+	client, err := net.Dial("tcp", "127.0.0.1:17767")
+	if err != nil {
+		fmt.Println("err:", err)
+		return
+	}
+	defer client.Close()
+
+	go func() {
+		input := make([]byte, 1024)
+		for {
+			n, err := os.Stdin.Read(input)
+			if err != nil {
+				fmt.Println("input err:", err)
+				continue
+			}
+			client.Write([]byte(input[:n]))
+		}
+	}()
+
+	buf := make([]byte, 1024)
+	for {
+		n, err := client.Read(buf)
+		if err != nil {
+			if err == io.EOF {
+				return
+			}
+			fmt.Println("read err:", err)
+			continue
+		}
+		fmt.Println(string(buf[:n]))
+
+	}
+}
+
 func main() {
 	var tcpServerPort int
 	var targetGrpcAddress string
@@ -14,6 +52,6 @@ func main() {
 	flag.Parse()
 	tcpServerAddress := fmt.Sprintf("0.0.0.0:%d", tcpServerPort)
 
-	tcp2grpcServer := proxy.NewTcp2GrpcServer(tcpServerAddress, targetGrpcAddress)
+	tcp2grpcServer := proxy.NewTCP2GrpcServer(tcpServerAddress, targetGrpcAddress)
 	tcp2grpcServer.Run()
 }

tools/tcp_grpc_proxy/cmd/tcpclient/main.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"flag"
+	"net"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+func main() {
+	var tcpServerAddress string
+	flag.StringVar(&tcpServerAddress, "tcp_server_address", "127.0.0.1:17767",
+		"TCP server address which the client connects to.")
+
+	conn, err := net.Dial("tcp", tcpServerAddress)
+	if err != nil {
+		logrus.Fatalf("Dail to tcp target %s error: %v", tcpServerAddress, err)
+	}
+	logrus.Infoln("Connected to", tcpServerAddress)
+	// Makes sure the connection gets closed
+	defer conn.Close()
+	defer logrus.Infoln("Connection closed to ", tcpServerAddress)
+
+	for {
+		conn.Write([]byte("hello world"))
+		logrus.Infof("Sent 'hello world' to server %s", tcpServerAddress)
+
+		tcpData := make([]byte, 64*1024)
+		_, err := conn.Read(tcpData)
+		if err != nil {
+			logrus.Fatalln("Read from tcp error: ", err)
+		}
+		logrus.Infof("Received '%s' from server", string(tcpData))
+
+		time.Sleep(time.Duration(5) * time.Second)
+	}
+}

tools/tcp_grpc_proxy/cmd/tcpserver/main.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"net"
+
+	"github.com/sirupsen/logrus"
+)
+
+func handleTCPConn(conn net.Conn) {
+	for {
+		tcpData := make([]byte, 64*1024)
+		bytesRead, err := conn.Read(tcpData)
+		if err != nil {
+			logrus.Fatalln("Read from tcp error: ", err)
+		}
+		logrus.Infof("TCP server got %d bytes", bytesRead)
+		conn.Write([]byte("This is a string from TCP server"))
+	}
+}
+
+func main() {
+	var tcpServerPort int
+	flag.IntVar(&tcpServerPort, "tcp_server_port", 17766, "TCP server port")
+	flag.Parse()
+	tcpServerAddress := fmt.Sprintf("0.0.0.0:%d", tcpServerPort)
+
+	listener, err := net.Listen("tcp", tcpServerAddress)
+	if err != nil {
+		logrus.Fatalln("Listen TCP error: ", err)
+	}
+	defer listener.Close()
+	logrus.Infoln("Run TCPServer at ", tcpServerAddress)
+
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			logrus.Errorln("TCP listener error:", err)
+			continue
+		}
+
+		logrus.Infoln("Got tcp connection")
+		go handleTCPConn(conn)
+	}
+}