Adjust cancel defer to ensure SQL completes Change-Id: I156a6aed08a0fb013c4ad63b596653648f4d70e6 Reviewed-on: https://skia-review.googlesource.com/c/buildbot/+/893056 Reviewed-by: Ashwin Verleker <ashwinpv@google.com> Commit-Queue: Tony Seaward <seawardt@google.com>
diff --git a/go/coverage/Makefile b/go/coverage/Makefile index fd501ec..dbc53ff 100644 --- a/go/coverage/Makefile +++ b/go/coverage/Makefile
@@ -20,7 +20,7 @@ run-coverage: $(BAZEL) build -c dbg //go/coverage/... ../../_bazel_bin/go/coverage/coverageserver/coverageserver_/coverageserver run \ - --config_filename=./config/demo.json \ + --config_filename=demo.json \ run-cockroachdb-emulators: ../../scripts/run_emulators/run_emulators start
diff --git a/go/coverage/coverageclient/BUILD.bazel b/go/coverage/coverageclient/BUILD.bazel index f8574b6..1fd75e9 100644 --- a/go/coverage/coverageclient/BUILD.bazel +++ b/go/coverage/coverageclient/BUILD.bazel
@@ -6,10 +6,15 @@ importpath = "go.skia.org/infra/go/coverage/coverageclient", visibility = ["//visibility:private"], deps = [ + "//go/auth", "//go/coverage/proto/v1", + "//go/skerr", "//go/sklog", "@org_golang_google_grpc//:go_default_library", + "@org_golang_google_grpc//credentials", "@org_golang_google_grpc//credentials/insecure", + "@org_golang_google_grpc//credentials/oauth", + "@org_golang_x_oauth2//google", ], )
diff --git a/go/coverage/coverageclient/main.go b/go/coverage/coverageclient/main.go index 81f3e67..c2123e0 100644 --- a/go/coverage/coverageclient/main.go +++ b/go/coverage/coverageclient/main.go
@@ -2,6 +2,7 @@ import ( "context" + "crypto/tls" "encoding/json" "flag" "io" @@ -9,12 +10,27 @@ "path/filepath" "strings" + "go.skia.org/infra/go/auth" pb "go.skia.org/infra/go/coverage/proto/v1" + "go.skia.org/infra/go/skerr" "go.skia.org/infra/go/sklog" + "golang.org/x/oauth2/google" "google.golang.org/grpc" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/credentials/oauth" ) +// getOauthCredentials returns a token source that will provide oauth tokens +// for the service account running the client process. +func getOauthCredentials(ctx context.Context) (oauth.TokenSource, error) { + tokenSource, err := google.DefaultTokenSource(ctx, auth.ScopeUserinfoEmail) + if err != nil { + return oauth.TokenSource{}, skerr.Wrapf(err, "Failed to create oauth token source.") + } + return oauth.TokenSource{TokenSource: tokenSource}, nil +} + // getGrpcConnection returns a ClientConn object that can be used to create individual // service clients for the coverage service. func getGrpcConnection(host string, insecure_conn bool) *grpc.ClientConn { @@ -23,6 +39,19 @@ if insecure_conn { opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) + } else { + tlsCreds := credentials.NewTLS(&tls.Config{ + // Since the communication is internal within the same GKE cluster, + // we do not need to verify the server certificate. + InsecureSkipVerify: true, + }) + tokenSource, err := getOauthCredentials(context.Background()) + if err != nil { + sklog.Errorf("Error Getting Token.") + return nil + + } + opts = append(opts, grpc.WithTransportCredentials(tlsCreds), grpc.WithPerRPCCredentials(tokenSource)) } conn, err := grpc.Dial(host, opts...) @@ -55,6 +84,7 @@ getAll := flag.Bool("getAll", false, "Get Full Database") host := flag.String("host", "localhost", "Hostname/IP of gRPC Service") port := flag.String("port", "8006", "Hostname/IP of gRPC Service") + insecure := flag.Bool("insecure", true, "Use TLS secure connection") flag.Parse() @@ -76,7 +106,7 @@ } rpcHost := *host + ":" + *port - conn := getGrpcConnection(rpcHost, true) + conn := getGrpcConnection(rpcHost, *insecure) client := pb.NewCoverageServiceClient(conn) if strings.HasPrefix(sampleFile, "get") {
diff --git a/go/coverage/coveragestore/sqlcoveragestore/sqlcoveragestore.go b/go/coverage/coveragestore/sqlcoveragestore/sqlcoveragestore.go index 861ab20..9bcd804 100644 --- a/go/coverage/coveragestore/sqlcoveragestore/sqlcoveragestore.go +++ b/go/coverage/coveragestore/sqlcoveragestore/sqlcoveragestore.go
@@ -75,44 +75,11 @@ }, nil } -func (s *CoverageStore) sqlExecUpdate(ctx context.Context, sqlStatement string, req *pb.CoverageChangeRequest) (int64, error) { - var result pgconn.CommandTag - var err error - var rows int64 - - for _, suite := range req.GetTestSuiteName() { - result, err = s.db.Exec(ctx, sqlStatement, req.GetFileName(), - req.GetBuilderName(), suite) - if err != nil { - sklog.Errorf("Update Failed") - return 0, err - } - rows += result.RowsAffected() - } - return rows, nil -} - -func (s *CoverageStore) sqlExecInsert(ctx context.Context, sqlStatement string, req *pb.CoverageChangeRequest) (int64, error) { - result, err := s.db.Exec(ctx, sqlStatement, req.GetFileName(), - req.GetBuilderName(), req.GetTestSuiteName()) - if err != nil { - return 0, err - } - return result.RowsAffected(), nil -} - -func (s *CoverageStore) sqlExecDelete(ctx context.Context, sqlStatement string, req *pb.CoverageChangeRequest) (int64, error) { - result, err := s.db.Exec(ctx, sqlStatement, req.GetFileName(), req.GetBuilderName()) - if err != nil { - return 0, err - } - return result.RowsAffected(), nil -} - // Add implements the coverage.CoverageStore interface. func (s *CoverageStore) Add(ctx context.Context, req *pb.CoverageChangeRequest) error { rows, err := s.sqlExecInsert(ctx, statements[addFile], req) if err != nil || rows > 0 { + sklog.Errorf("Add Failed: %s", statements[addFile]) return err } rows, err = s.sqlExecInsert(ctx, statements[addBuilder], req) @@ -204,3 +171,37 @@ sklog.Debugf("Responses: %s", responses) return responses, err } + +func (s *CoverageStore) sqlExecUpdate(ctx context.Context, sqlStatement string, req *pb.CoverageChangeRequest) (int64, error) { + var result pgconn.CommandTag + var err error + var rows int64 + + for _, suite := range req.GetTestSuiteName() { + result, err = s.db.Exec(ctx, sqlStatement, req.GetFileName(), + req.GetBuilderName(), suite) + if err != nil { + sklog.Errorf("Update Failed") + return 0, err + } + rows += result.RowsAffected() + } + return rows, nil +} + +func (s *CoverageStore) sqlExecInsert(ctx context.Context, sqlStatement string, req *pb.CoverageChangeRequest) (int64, error) { + result, err := s.db.Exec(ctx, sqlStatement, req.GetFileName(), + req.GetBuilderName(), req.GetTestSuiteName()) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +func (s *CoverageStore) sqlExecDelete(ctx context.Context, sqlStatement string, req *pb.CoverageChangeRequest) (int64, error) { + result, err := s.db.Exec(ctx, sqlStatement, req.GetFileName(), req.GetBuilderName()) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +}
diff --git a/go/coverage/service/service_impl.go b/go/coverage/service/service_impl.go index ce50be8..af99f0c 100644 --- a/go/coverage/service/service_impl.go +++ b/go/coverage/service/service_impl.go
@@ -43,23 +43,25 @@ } } -func checkContextDeadline(ctx context.Context) context.Context { +func checkContextDeadline(ctx context.Context) (context.Context, context.CancelFunc) { _, status := ctx.Deadline() if status { - return ctx + return ctx, nil } ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second) - defer cancel() - return ctxWithTimeout + return ctxWithTimeout, cancel } // Checks file/builder pair from Database and returns available test suites. func (s *coverageService) GetAllFiles(ctx context.Context, req *pb.CoverageRequest) (*pb.CoverageAllResponses, error) { - ctx = checkContextDeadline(ctx) + updatedCtx, cancel := checkContextDeadline(ctx) + results, err := s.coverageStore.ListAll(updatedCtx, req) + if cancel != nil { + defer cancel() + } - results, err := s.coverageStore.ListAll(ctx, req) if err != nil { - sklog.Errorf("GetTestSuite Failed: %v with error: %v", req, err) + sklog.Errorf("GetAllSuite Failed: %v with error: %v", req, err) } var allResults pb.CoverageAllResponses allResults.Responses = results @@ -68,9 +70,11 @@ // Checks file/builder pair from Database and returns available test suites. func (s *coverageService) GetTestSuite(ctx context.Context, req *pb.CoverageListRequest) (*pb.CoverageListResponse, error) { - ctx = checkContextDeadline(ctx) - - test_suites, err := s.coverageStore.List(ctx, req) + updatedCtx, cancel := checkContextDeadline(ctx) + test_suites, err := s.coverageStore.List(updatedCtx, req) + if cancel != nil { + defer cancel() + } status := successStatus if err != nil || test_suites == nil { @@ -84,9 +88,12 @@ // Inserts file/builder pair from Database. func (s *coverageService) InsertFile(ctx context.Context, req *pb.CoverageChangeRequest) (*pb.CoverageChangeResponse, error) { - ctx = checkContextDeadline(ctx) + updatedCtx, cancel := checkContextDeadline(ctx) + err := s.coverageStore.Add(updatedCtx, req) + if cancel != nil { + defer cancel() + } - err := s.coverageStore.Add(ctx, req) status := successStatus message := "Add Successful" @@ -105,9 +112,12 @@ // Deletes file/builder pair from Database. func (s *coverageService) DeleteFile(ctx context.Context, req *pb.CoverageChangeRequest) (*pb.CoverageChangeResponse, error) { - ctx = checkContextDeadline(ctx) + updatedCtx, cancel := checkContextDeadline(ctx) + err := s.coverageStore.Delete(updatedCtx, req) + if cancel != nil { + defer cancel() + } - err := s.coverageStore.Delete(ctx, req) status := successStatus message := "Delete Successful"