Remove GRPC Proxy for Log Rest API Call

This reduces the need for multiple translation done by proxy and
improves the performance.

cmd/api/main.go

@@ -297,33 +297,30 @@ func main() {
 	ctx := context.Background()
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	httpMux := runtime.NewServeMux(serverMuxOptions...)
+	var httpMux http.Handler
+	httpMux = runtime.NewServeMux(serverMuxOptions...)
 	opts := []grpc.DialOption{
 		grpc.WithTransportCredentials(creds),
 		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(100 * 1024 * 1024)),
 		grpc.WithNoProxy(),
 	}
 
 	// Register gRPC server endpoint to gRPC gateway
-	err = v1alpha2pb.RegisterResultsHandlerFromEndpoint(ctx, httpMux, ":"+serverConfig.SERVER_PORT, opts)
+	err = v1alpha2pb.RegisterResultsHandlerFromEndpoint(ctx, httpMux.(*runtime.ServeMux), ":"+serverConfig.SERVER_PORT, opts)
 	if err != nil {
 		log.Fatal("Error registering gRPC server endpoint for Results API: ", err)
 	}
 
 	if serverConfig.LOGS_API && !v1a2.LogPluginServer.IsLogPluginEnabled {
 		log.Info("Registering server endpoints for Logs v1alpha2 API")
-		err = v1alpha2pb.RegisterLogsHandlerFromEndpoint(ctx, httpMux, ":"+serverConfig.SERVER_PORT, opts)
+		err = v1alpha2pb.RegisterLogsHandlerFromEndpoint(ctx, httpMux.(*runtime.ServeMux), ":"+serverConfig.SERVER_PORT, opts)
 		if err != nil {
 			log.Fatal("Error registering gRPC server endpoints for Logs v1alpha2 API: ", err)
 		}
-	} else if serverConfig.LOGS_API && v1a2.LogPluginServer.IsLogPluginEnabled {
-		log.Info("Registering server endpoints for Logs v1alpha3 API")
-		err = v1alpha3pb.RegisterLogsHandlerFromEndpoint(ctx, httpMux, ":"+serverConfig.SERVER_PORT, opts)
-		if err != nil {
-			log.Fatal("Error registering gRPC server endpoints for Logs v1alpha3 API: ", err)
-		}
 	}
 
+	httpMux = v1alpha2.Handler(httpMux, v1a2.LogPluginServer)
+
 	// Start server with gRPC and REST handler
 	log.Infof("gRPC and REST server listening on: %s", serverConfig.SERVER_PORT)
 	if tlsError != nil {

pkg/api/server/v1alpha2/handlers.go

@@ -0,0 +1,16 @@
+package server
+
+import (
+	"net/http"
+)
+
+// Handler returns a http.Handler that serves the gRPC server and the log plugin server
+func Handler(grpcMux http.Handler, pluginServer *LogPluginServer) http.Handler {
+	mux := http.NewServeMux()
+	mux.Handle("/", grpcMux)
+	if pluginServer != nil && pluginServer.IsLogPluginEnabled {
+		mux.Handle("/apis/results.tekton.dev/v1alpha2/parents/{parent}/results/{resultID}/logs/{recordID}", pluginServer.LogMux())
+		mux.Handle("/apis/results.tekton.dev/v1alpha3/parents/{parent}/results/{resultID}/logs/{recordID}", pluginServer.LogMux())
+	}
+	return mux
+}

pkg/api/server/v1alpha2/plugin_logs.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
 	"github.com/tektoncd/results/pkg/api/server/db"
@@ -61,6 +62,7 @@ func (s *LogPluginServer) GetLog(req *pb3.GetLogRequest, srv pb3.Logs_GetLogServ
 	if err != nil {
 		s.logger.Error(err)
 	}
+
 	_, err = writer.Flush()
 	if err != nil {
 		s.logger.Error(err)
@@ -69,7 +71,7 @@ func (s *LogPluginServer) GetLog(req *pb3.GetLogRequest, srv pb3.Logs_GetLogServ
 	return nil
 }
 
-func (s *LogPluginServer) getPluginLogs(writer *logs.BufferedLog, parent string, rec *db.Record) error {
+func (s *LogPluginServer) getPluginLogs(writer io.Writer, parent string, rec *db.Record) error {
 	switch strings.ToLower(s.config.LOGS_TYPE) {
 	case string(v1alpha3.LokiLogType):
 		return s.getLokiLogs(writer, parent, rec)
@@ -79,7 +81,7 @@ func (s *LogPluginServer) getPluginLogs(writer *logs.BufferedLog, parent string,
 	}
 }
 
-func (s *LogPluginServer) getLokiLogs(writer *logs.BufferedLog, parent string, rec *db.Record) error {
+func (s *LogPluginServer) getLokiLogs(writer io.Writer, parent string, rec *db.Record) error {
 	URL, err := url.Parse(s.config.LOGGING_PLUGIN_API_URL)
 	if err != nil {
 		s.logger.Error(err)
@@ -247,3 +249,32 @@ func (s *LogPluginServer) getLokiLogs(writer *logs.BufferedLog, parent string, r
 	return nil
 
 }
+
+// LogMux returns a http.Handler that serves the log plugin server
+func (s *LogPluginServer) LogMux() http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// TODO: Create a new log handler
+		ctx := r.Context()
+		ctx = metadata.AppendToOutgoingContext(ctx, "authorization", r.Header.Get("Authorization"))
+		parent := r.PathValue("parent")
+		if err := s.auth.Check(ctx, parent, auth.ResourceLogs, auth.PermissionGet); err != nil {
+			s.logger.Error(err)
+			http.Error(w, "Not Authorized", http.StatusUnauthorized)
+			return
+		}
+		recID := r.PathValue("recordID")
+		res := r.PathValue("resultID")
+		s.logger.Infof("recordID: %s resultID: %s name: %s", recID, res, parent)
+		rec, err := getRecord(s.db, parent, res, recID)
+		if err != nil {
+			s.logger.Error(err)
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+
+		err = s.getPluginLogs(w, parent, rec)
+		if err != nil {
+			s.logger.Error(err)
+			http.Error(w, "Failed to stream logs err: "+err.Error(), http.StatusInternalServerError)
+		}
+	})
+}

proto/v1alpha3/log.proto

@@ -16,7 +16,6 @@ syntax = "proto3";
 
 package tekton.results.v1alpha3;
 
-import "google/api/annotations.proto";
 import "google/api/field_behavior.proto";
 import "google/api/client.proto";
 import "google/api/resource.proto";
@@ -27,9 +26,6 @@ option go_package = "github.com/tektoncd/results/proto/v1alpha3/results_go_proto
 
 service Logs {
   rpc GetLog(GetLogRequest) returns (stream google.api.HttpBody) {
-    option (google.api.http) = {
-      get: "/apis/results.tekton.dev/v1alpha2/parents/{name=*/results/*/logs/*}"
-    };
     option (google.api.method_signature) = "name";
   }
 }