lostluck commented on code in PR #28893: URL: https://github.com/apache/beam/pull/28893#discussion_r1367770384
########## .test-infra/mock-apis/src/main/go/internal/cache/cache.go: ########## @@ -0,0 +1,122 @@ +// 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 cache + +import ( + "context" + "errors" + "fmt" + "log/slog" + "reflect" + "time" + + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/logging" +) + +var ( + + // ErrNotExist is an error indicating that a resource does not exist + ErrNotExist = errors.New("resource does not exist") +) + +// IsNotExist is true when err is ErrNotExist. +func IsNotExist(err error) bool { + return errors.Is(err, ErrNotExist) +} + +// Options for running the Refresher. +type Options struct { + Setter UInt64Setter + Logger *slog.Logger +} + +// Refresher refreshes a value in a cache on a set interval. +type Refresher struct { + opts *Options + stop chan struct{} +} + +// NewRefresher instantiates a Refresher. +func NewRefresher(ctx context.Context, opts *Options) (*Refresher, error) { + if opts.Logger == nil { + opts.Logger = logging.New(&logging.Options{ + Name: reflect.TypeOf((*Refresher)(nil)).PkgPath(), + }) + } + + if opts.Setter == nil { + return nil, fmt.Errorf("%T.Setter is nil but required", opts) + } + + if err := opts.Setter.Alive(ctx); err != nil { + return nil, err + } + + ref := &Refresher{ + opts: opts, + } + + return ref, nil +} + +// Stop the Refresher. +func (ref *Refresher) Stop() { + ref.stop <- struct{}{} +} + +// Refresh the size of the associated key at an interval. +func (ref *Refresher) Refresh(ctx context.Context, key string, size uint64, interval time.Duration) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + ref.stop = make(chan struct{}) + attrs := []slog.Attr{ + { + Key: "key", + Value: slog.StringValue(key), + }, + { + Key: "size", + Value: slog.Uint64Value(size), + }, + { + Key: "interval", + Value: slog.StringValue(interval.String()), + }, + } + + ref.opts.Logger.LogAttrs(ctx, slog.LevelInfo, "starting refresher service", attrs...) Review Comment: Instead of using attrs... on every call, use the logger.With method early on to get a logger with those outputs built. https://pkg.go.dev/log/slog#Logger.With ########## .test-infra/mock-apis/src/main/go/internal/cache/cache.go: ########## @@ -0,0 +1,122 @@ +// 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 cache + +import ( + "context" + "errors" + "fmt" + "log/slog" + "reflect" + "time" + + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/logging" +) + +var ( + Review Comment: rm blank line ########## .test-infra/mock-apis/src/main/go/internal/service/echo/echo.go: ########## @@ -0,0 +1,185 @@ +// 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 echo contains the EchoService API implementation. +package echo + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "path" + "reflect" + "time" + + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/cache" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/logging" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/metric" + echov1 "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/proto/echo/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/status" +) + +const ( + metricsNamePrefix = "echo" + echoPath = "/proto.echo.v1.EchoService/Echo" + echoPathAlias = "/v1/echo" + healthPath = "/grpc.health.v1.Health/Check" + healthPathAlias = "/v1/healthz" +) + +type Options struct { + Decrementer cache.Decrementer + MetricsWriter metric.Writer + Logger *slog.Logger + LoggingAttrs []slog.Attr +} + +// Register a grpc.Server with the echov1.EchoService. Returns a http.Handler or error. +func Register(s *grpc.Server, opts *Options) (http.Handler, error) { + if opts.Logger == nil { + opts.Logger = logging.New(&logging.Options{ + Name: reflect.TypeOf((*echo)(nil)).PkgPath(), + }) + } + srv := &echo{ + opts: opts, + } + + echov1.RegisterEchoServiceServer(s, srv) + grpc_health_v1.RegisterHealthServer(s, srv) + + return srv, nil +} + +type echo struct { + echov1.UnimplementedEchoServiceServer + grpc_health_v1.UnimplementedHealthServer + opts *Options +} + +// ServeHTTP implements http.Handler, allowing echo to support HTTP clients in addition to gRPC. +func (srv *echo) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case echoPath, echoPathAlias: + srv.httpHandler(w, r) + case healthPath, healthPathAlias: + srv.checkHandler(w, r) + default: + http.Error(w, fmt.Sprintf("%s not found", r.URL.Path), http.StatusNotFound) + } +} + +// Check checks whether echo service's underlying decrementer is alive. +func (srv *echo) Check(ctx context.Context, _ *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { + if err := srv.opts.Decrementer.Alive(ctx); err != nil { + return nil, err + } + return &grpc_health_v1.HealthCheckResponse{ + Status: grpc_health_v1.HealthCheckResponse_SERVING, + }, nil +} + +func (srv *echo) checkHandler(w http.ResponseWriter, r *http.Request) { + resp, err := srv.Check(r.Context(), nil) + if err != nil { + + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if err := json.NewEncoder(w).Encode(resp); err != nil { + srv.opts.Logger.LogAttrs(context.Background(), slog.LevelError, err.Error(), srv.opts.LoggingAttrs...) Review Comment: Instead of using context.Background() for these logging calls, use the one attached to the request: https://pkg.go.dev/net/http#Request.Context ########## .test-infra/mock-apis/src/main/go/cmd/service/echo/main.go: ########## @@ -0,0 +1,148 @@ +// 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. + +// echo is an executable that runs the echov1.EchoService. +package main + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/http" + "os" + "os/signal" + + gcplogging "cloud.google.com/go/logging" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/cache" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/environment" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/logging" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/service/echo" + "github.com/redis/go-redis/v9" + "google.golang.org/grpc" +) + +var ( + env = []environment.Variable{ + environment.CacheHost, + environment.GrpcPort, + environment.HttpPort, + } + + logger *slog.Logger + logAttrs []slog.Attr + opts = &logging.Options{ + Name: "echo", + } +) + +func init() { + for _, v := range env { + logAttrs = append(logAttrs, slog.Attr{ + Key: v.Key(), + Value: slog.StringValue(v.Value()), + }) + } +} + +func main() { + ctx := context.Background() + + if !environment.ProjectId.Missing() { + client, err := gcplogging.NewClient(ctx, environment.ProjectId.Value()) + if err != nil { + slog.LogAttrs(ctx, slog.LevelError, err.Error(), logAttrs...) + os.Exit(1) + } + + opts.Client = client + } + + logger = logging.New(opts) + + if err := run(ctx); err != nil { + logger.LogAttrs(ctx, slog.LevelError, err.Error(), logAttrs...) + os.Exit(1) + } +} + +func run(ctx context.Context) error { + ctx, cancel := signal.NotifyContext(ctx, os.Interrupt) + defer cancel() + + if err := environment.Missing(env...); err != nil { + return err + } + + grpcPort, err := environment.GrpcPort.Int() + if err != nil { + return err + } + grpcAddress := fmt.Sprintf(":%v", grpcPort) + + httpPort, err := environment.HttpPort.Int() + if err != nil { + return err + } + httpAddress := fmt.Sprintf(":%v", httpPort) + + s := grpc.NewServer() + defer s.GracefulStop() + + r := redis.NewClient(&redis.Options{ + Addr: environment.CacheHost.Value(), + }) + + echoOpts := &echo.Options{ + Decrementer: (*cache.RedisCache)(r), + LoggingAttrs: logAttrs, + Logger: logger, Review Comment: If the intent here is to always print out these attrs with every request, just build them into the logger instead of forcing per-callsite behavior on yourself. https://pkg.go.dev/log/slog#Logger.With If the attributes are only going to be logged sometimes though, document the field with the conditions they're logged, so it's clear what that field is for. ########## .test-infra/mock-apis/src/main/go/internal/metric/doc.go: ########## @@ -0,0 +1,17 @@ +// 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 metric supports monitoring. +package metric Review Comment: Style note: If the documentation is small, and the package is small (file wise), it's better to just put it on top of one ot the non test files. A separate doc file works best when the package documentation is long, or doesn't naturally fit into one of the existing files. Usually though in that case there'd be a file named the same as the package. If it were me, I'd rename interface.go to metric.go since it's an evident entry point to the package. ########## .test-infra/mock-apis/src/main/java/org/apache/beam/testinfra/mockapis/echo/v1/Echo.java: ########## @@ -0,0 +1,1447 @@ +/* + * 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 org.apache.beam.testinfra.mockapis.echo.v1; Review Comment: Question: Is all this java, generated code? ########## .test-infra/mock-apis/src/main/go/internal/service/echo/echo.go: ########## @@ -0,0 +1,185 @@ +// 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 echo contains the EchoService API implementation. +package echo + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "path" + "reflect" + "time" + + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/cache" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/logging" + "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/metric" + echov1 "github.com/apache/beam/test-infra/mock-apis/src/main/go/internal/proto/echo/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/status" +) + +const ( + metricsNamePrefix = "echo" + echoPath = "/proto.echo.v1.EchoService/Echo" + echoPathAlias = "/v1/echo" + healthPath = "/grpc.health.v1.Health/Check" + healthPathAlias = "/v1/healthz" +) + +type Options struct { + Decrementer cache.Decrementer + MetricsWriter metric.Writer + Logger *slog.Logger + LoggingAttrs []slog.Attr +} + +// Register a grpc.Server with the echov1.EchoService. Returns a http.Handler or error. +func Register(s *grpc.Server, opts *Options) (http.Handler, error) { + if opts.Logger == nil { + opts.Logger = logging.New(&logging.Options{ Review Comment: Same comment here WRT the logger and the always used LoggingAttrs. Use `.With()` to build them into the logger. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
