[BEAM-12983] [Playground] Initial gRPC server

playground/backend/cmd/server/controller.go

@@ -0,0 +1,55 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package main
+
+import (
+	pb "beam.apache.org/playground/backend/internal/api"
+	"context"
+	"github.com/google/uuid"
+)
+
+type playgroundController struct {
+	pb.UnimplementedPlaygroundServiceServer
+}
+
+//RunCode is running code from requests using a particular SDK
+func (controller *playgroundController) RunCode(ctx context.Context, info *pb.RunCodeRequest) (*pb.RunCodeResponse, error) {
+	// TODO implement this method
+	pipelineInfo := pb.RunCodeResponse{PipelineUuid: uuid.NewString()}
+
+	return &pipelineInfo, nil
+}
+
+//CheckStatus is checking status for the specific pipeline by PipelineUuid
+func (controller *playgroundController) CheckStatus(ctx context.Context, info *pb.CheckStatusRequest) (*pb.CheckStatusResponse, error) {
+	// TODO implement this method
+	status := pb.CheckStatusResponse{Status: pb.Status_STATUS_FINISHED}
+	return &status, nil
+}
+
+//GetRunOutput is returning output of execution for specific pipeline by PipelineUuid
+func (controller *playgroundController) GetRunOutput(ctx context.Context, info *pb.GetRunOutputRequest) (*pb.GetRunOutputResponse, error) {
+	// TODO implement this method
+	pipelineResult := pb.GetRunOutputResponse{Output: "Test Pipeline Result"}
+
+	return &pipelineResult, nil
+}
+
+//GetCompileOutput is returning output of compilation for specific pipeline by PipelineUuid
+func (controller *playgroundController) GetCompileOutput(ctx context.Context, info *pb.GetCompileOutputRequest) (*pb.GetCompileOutputResponse, error) {
+	// TODO implement this method
+	compileOutput := pb.GetCompileOutputResponse{Output: "test compile output"}
+	return &compileOutput, nil
+}

playground/backend/cmd/server/controller_test.go

@@ -0,0 +1,128 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package main
+
+import (
+	pb "beam.apache.org/playground/backend/internal/api"
+	"context"
+	"github.com/google/uuid"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/test/bufconn"
+	"log"
+	"net"
+	"testing"
+)
+
+const bufSize = 1024 * 1024
+
+var lis *bufconn.Listener
+
+func TestMain(m *testing.M) {
+	server := setup()
+	defer teardown(server)
+	m.Run()
+}
+
+func setup() *grpc.Server {
+	lis = bufconn.Listen(bufSize)
+	s := grpc.NewServer()
+	pb.RegisterPlaygroundServiceServer(s, &playgroundController{})
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("Server exited with error: %v", err)
+		}
+	}()
+	return s
+}
+
+func teardown(server *grpc.Server) {
+	server.Stop()
+}
+
+func bufDialer(context.Context, string) (net.Conn, error) {
+	return lis.Dial()
+}
+func TestPlaygroundController_RunCode(t *testing.T) {
+	ctx := context.Background()
+	conn, err := grpc.DialContext(ctx, "bufnet", grpc.WithContextDialer(bufDialer), grpc.WithInsecure())
+	if err != nil {
+		t.Fatalf("Failed to dial bufnet: %v", err)
+	}
+	defer conn.Close()
+	client := pb.NewPlaygroundServiceClient(conn)
+	code := pb.RunCodeRequest{
+		Code: "test",
+		Sdk:  pb.Sdk_SDK_JAVA,
+	}
+	pipelineMeta, err := client.RunCode(ctx, &code)
+	if err != nil {
+		t.Fatalf("runCode failed: %v", err)
+	}
+	log.Printf("Response: %+v", pipelineMeta)
+}
+
+func TestPlaygroundController_CheckStatus(t *testing.T) {
+	ctx := context.Background()
+	conn, err := grpc.DialContext(ctx, "bufnet", grpc.WithContextDialer(bufDialer), grpc.WithInsecure())
+	if err != nil {
+		t.Fatalf("Failed to dial bufnet: %v", err)
+	}
+	defer conn.Close()
+	client := pb.NewPlaygroundServiceClient(conn)
+	pipelineMeta := pb.CheckStatusRequest{
+		PipelineUuid: uuid.NewString(),
+	}
+	status, err := client.CheckStatus(ctx, &pipelineMeta)
+	if err != nil {
+		t.Fatalf("runCode failed: %v", err)
+	}
+	log.Printf("Response: %+v", status)
+}
+
+func TestPlaygroundController_GetCompileOutput(t *testing.T) {
+	ctx := context.Background()
+	conn, err := grpc.DialContext(ctx, "bufnet", grpc.WithContextDialer(bufDialer), grpc.WithInsecure())
+	if err != nil {
+		t.Fatalf("Failed to dial bufnet: %v", err)
+	}
+	defer conn.Close()
+	client := pb.NewPlaygroundServiceClient(conn)
+	pipelineMeta := pb.GetCompileOutputRequest{
+		PipelineUuid: uuid.NewString(),
+	}
+	compileOutput, err := client.GetCompileOutput(ctx, &pipelineMeta)
+	if err != nil {
+		t.Fatalf("runCode failed: %v", err)
+	}
+	log.Printf("Response: %+v", compileOutput)
+}
+
+func TestPlaygroundController_GetRunOutput(t *testing.T) {
+	ctx := context.Background()
+	conn, err := grpc.DialContext(ctx, "bufnet", grpc.WithContextDialer(bufDialer), grpc.WithInsecure())
+	if err != nil {
+		t.Fatalf("Failed to dial bufnet: %v", err)
+	}
+	defer conn.Close()
+	client := pb.NewPlaygroundServiceClient(conn)
+	pipelineMeta := pb.GetRunOutputRequest{
+		PipelineUuid: uuid.NewString(),
+	}
+	runOutput, err := client.GetRunOutput(ctx, &pipelineMeta)
+	if err != nil {
+		t.Fatalf("runCode failed: %v", err)
+	}
+	log.Printf("Response: %+v", runOutput)
+}

playground/backend/cmd/server/http.go

@@ -0,0 +1,37 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package main
+
+import (
+	"beam.apache.org/playground/backend/internal/environment"
+	"context"
+	"google.golang.org/grpc/grpclog"
+	"net/http"
+)
+
+// listenHttp binds the http.Handler on the TCP network address
+func listenHttp(ctx context.Context, errChan chan error, envs environment.ServerEnvs, handler http.Handler) {
+	grpclog.Infof("listening HTTP at %s\n", envs.Address())
+	if err := http.ListenAndServe(envs.Address(), handler); err != nil {
+		errChan <- err
+		return
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		}
+	}
+}

playground/backend/cmd/server/server.go

@@ -16,9 +16,57 @@
 package main
 
 import (
-	"beam.apache.org/playground/backend/internal/executors"
+	pb "beam.apache.org/playground/backend/internal/api"
+	"beam.apache.org/playground/backend/internal/environment"
+	"context"
+	"github.com/improbable-eng/grpc-web/go/grpcweb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/grpclog"
+	"log"
+	"os"
 )
 
+// runServer is starting http server wrapped on grpc
+func runServer() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	envService := environment.NewEnvironment()
+	grpcServer := grpc.NewServer()
+	pb.RegisterPlaygroundServiceServer(grpcServer, &playgroundController{})
+
+	grpclog.SetLoggerV2(grpclog.NewLoggerV2(os.Stdout, os.Stdout, os.Stderr))
+	handler := Wrap(grpcServer, getGrpcWebOptions())
+	errChan := make(chan error)
+
+	go listenHttp(ctx, errChan, envService.ServerEnvs, handler)
+
+	for {
+		select {
+		case err := <-errChan:
+			return err
+		case <-ctx.Done():
+			log.Println("interrupt signal received; stopping...")
+			return nil
+		}
+	}
+}
+
+// getGrpcWebOptions returns grpcweb options needed to configure wrapper
+func getGrpcWebOptions() []grpcweb.Option {
+	return []grpcweb.Option{
+		grpcweb.WithCorsForRegisteredEndpointsOnly(false),
+		grpcweb.WithAllowNonRootResource(true),
+		grpcweb.WithOriginFunc(func(origin string) bool {
+			return true
+		}),
+	}
+
+}
+
 func main() {
-	_ = executors.GoExecutor{}
+	err := runServer()
+	if err != nil {
+		panic(err)
+	}
 }

playground/backend/cmd/server/wrapper.go

@@ -0,0 +1,40 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package main
+
+import (
+	"github.com/improbable-eng/grpc-web/go/grpcweb"
+	"github.com/rs/cors"
+	"google.golang.org/grpc"
+	"net/http"
+)
+
+// Wrap a grpc.Server as a http.Handler
+func Wrap(svc *grpc.Server, options []grpcweb.Option) http.Handler {
+	wrappedServer := grpcweb.WrapServer(svc, options...)
+	// TODO address what is the minimum necessary
+	cors.AllowAll().Handler(&wrapper{
+		svc: wrappedServer,
+	})
+	return wrappedServer
+}
+
+type wrapper struct {
+	svc *grpcweb.WrappedGrpcServer
+}
+
+func (wrap *wrapper) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
+	wrap.svc.ServeHTTP(resp, req)
+}

playground/backend/go.mod

@@ -19,6 +19,8 @@ go 1.16
 
 require (
 	github.com/google/uuid v1.3.0
+	github.com/improbable-eng/grpc-web v0.14.1
+	github.com/rs/cors v1.8.0
 	google.golang.org/grpc v1.41.0
 	google.golang.org/protobuf v1.27.1
 )