Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[accountingservice] adds otel logging support #1477

Closed
wants to merge 6 commits into from
Closed

[accountingservice] adds otel logging support #1477

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
230b09a
Added otellogrus to the accounting service
Kimbohlovette Mar 23, 2024
ec2ddb4
Merge branch 'open-telemetry:main' into Add-otel-log-support-to-accou…
Kimbohlovette Mar 23, 2024
f50deea
propagated the context from sarama consumer group session into the lo…
Kimbohlovette Mar 23, 2024
6a5545a
Replaced logrus with slog
Kimbohlovette Mar 28, 2024
7c016d3
Merge branch 'main' into Add-otel-log-support-to-accountingservice-1/7
Kimbohlovette Mar 28, 2024
d138dc8
Applied request changes/ initialized slog in main and removed initLog…
Kimbohlovette Mar 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/accountingservice/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,14 @@ require (
github.com/klauspost/compress v1.17.5 // indirect
github.com/pierrec/lz4/v4 v4.1.21 // indirect
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
github.com/uptrace/opentelemetry-go-extra/otellogrus v0.2.4 // indirect
github.com/uptrace/opentelemetry-go-extra/otelutil v0.2.4 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.24.0 // indirect
go.opentelemetry.io/otel/metric v1.24.0 // indirect
go.opentelemetry.io/proto/otlp v1.1.0 // indirect
golang.org/x/crypto v0.19.0 // indirect
golang.org/x/net v0.21.0 // indirect
golang.org/x/sys v0.17.0 // indirect
golang.org/x/sys v0.18.0 // indirect
golang.org/x/text v0.14.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20240125205218-1f4bbc51befe // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe // indirect
Expand Down
6 changes: 6 additions & 0 deletions src/accountingservice/go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,10 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/uptrace/opentelemetry-go-extra/otellogrus v0.2.4 h1:3Ok5iuhXYDM2cfs9/a5GPsWma58uo8FJhjWndQtmt/A=
github.com/uptrace/opentelemetry-go-extra/otellogrus v0.2.4/go.mod h1:EiUD7/t6Vre9nayvP5IVdOqbLMabf/c4k8H2j1yHhdg=
github.com/uptrace/opentelemetry-go-extra/otelutil v0.2.4 h1:A6+6ZGgLRoUTD+Jkw/Ph0g8HKiHUsiGlbngcSqBaHsw=
github.com/uptrace/opentelemetry-go-extra/otelutil v0.2.4/go.mod h1:gNYQe4RRVyszriFOhuMpwpAu4kdoFlZgcsw6dcIDFWg=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
go.opentelemetry.io/otel v1.24.0 h1:0LAOdjNmQeSTzGBzduGe/rU4tZhMwL5rWgtp9Ku5Jfo=
go.opentelemetry.io/otel v1.24.0/go.mod h1:W7b9Ozg4nkF5tWI5zsXkaKKDjdVjpD4oAt9Qi/MArHo=
Expand Down Expand Up @@ -114,6 +118,8 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
Expand Down
14 changes: 5 additions & 9 deletions src/accountingservice/kafka/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@ package kafka

import (
"context"
"log/slog"

pb "github.com/open-telemetry/opentelemetry-demo/src/accountingservice/genproto/oteldemo"

"github.com/IBM/sarama"
"github.com/sirupsen/logrus"
"google.golang.org/protobuf/proto"
)

Expand All @@ -17,7 +18,7 @@ var (
GroupID = "accountingservice"
)

func StartConsumerGroup(ctx context.Context, brokers []string, log *logrus.Logger) (sarama.ConsumerGroup, error) {
func StartConsumerGroup(ctx context.Context, brokers []string, log *slog.Logger) (sarama.ConsumerGroup, error) {
saramaConfig := sarama.NewConfig()
saramaConfig.Version = ProtocolVersion
// So we can know the partition and offset of messages.
Expand All @@ -42,7 +43,7 @@ func StartConsumerGroup(ctx context.Context, brokers []string, log *logrus.Logge
}

type groupHandler struct {
log *logrus.Logger
log *slog.Logger
}

func (g *groupHandler) Setup(_ sarama.ConsumerGroupSession) error {
Expand All @@ -62,12 +63,7 @@ func (g *groupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim s
if err != nil {
return err
}

g.log.WithFields(logrus.Fields{
"orderId": orderResult.OrderId,
"messageTimestamp": message.Timestamp,
"messageTopic": message.Topic,
}).Info("Message claimed")
g.log.LogAttrs(session.Context(), slog.LevelInfo, "Message claimed", slog.String("orderId", orderResult.OrderId), slog.String("messageTimestamp", message.Timestamp.String()), slog.String("messageTopic", message.Topic))
Copy link
Member

@pellared pellared Apr 2, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Notice that session.Context() would NOT contain the trace context information as NewOTelInterceptor does not pass the context created in oi.tracer.Start. Therefore, the logs would not contain the TraceID and SpanID fields.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, so should I just use a new context like context.TODO()?

Copy link
Member

@pellared pellared Apr 2, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You should try to refine the sarama instrumentation so that you can propagate trace context to the logs.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay

session.MarkMessage(message, "")

case <-session.Context().Done():
Expand Down
43 changes: 17 additions & 26 deletions src/accountingservice/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,14 @@ package main
import (
"context"
"fmt"
"log/slog"
"os"
"os/signal"
"strings"
"sync"
"syscall"
"time"

"github.com/IBM/sarama"
"github.com/sirupsen/logrus"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/propagation"
Expand All @@ -27,24 +26,14 @@ import (
"github.com/open-telemetry/opentelemetry-demo/src/accountingservice/kafka"
)

var log *logrus.Logger
var resource *sdkresource.Resource
var initResourcesOnce sync.Once

func init() {
log = logrus.New()
log.Level = logrus.DebugLevel
log.Formatter = &logrus.JSONFormatter{
FieldMap: logrus.FieldMap{
logrus.FieldKeyTime: "timestamp",
logrus.FieldKeyLevel: "severity",
logrus.FieldKeyMsg: "message",
},
TimestampFormat: time.RFC3339Nano,
}
log.Out = os.Stdout
func initLogger() *slog.Logger {
logger := slog.New(slog.NewJSONHandler(os.Stderr, nil)).With("service", "accounting")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the place where you would use to otelslog bridge instead of slog.NewJSONHandle.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Noted

slog.SetDefault(logger)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are you setting the default logger. You never use it afterwards.

Consider removing this line or not returning the logger in this function and using slog.LogAttrs instead of logger.LogAttrs.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay I will do the changes as requested
Thank you for the reviews 😊

return logger
}

func initResource() *sdkresource.Resource {
initResourcesOnce.Do(func() {
extraResources, _ := sdkresource.New(
Expand Down Expand Up @@ -79,39 +68,41 @@ func initTracerProvider() (*sdktrace.TracerProvider, error) {
}

func main() {
logger := initLogger()
ctx := context.Background()
tp, err := initTracerProvider()
if err != nil {
log.Fatal(err)
logger.LogAttrs(ctx, slog.LevelError, "failed to initialize trace provider", slog.String("error", err.Error()))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
logger.LogAttrs(ctx, slog.LevelError, "failed to initialize trace provider", slog.String("error", err.Error()))
logger.LogAttrs(ctx, slog.LevelError, "Failed to initialize trace provider", slog.String("error", err.Error()))

}
defer func() {
if err := tp.Shutdown(context.Background()); err != nil {
log.Printf("Error shutting down tracer provider: %v", err)
if err := tp.Shutdown(ctx); err != nil {
logger.LogAttrs(ctx, slog.LevelError, "failed to shotdown properly", slog.String("error", err.Error()))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
logger.LogAttrs(ctx, slog.LevelError, "failed to shotdown properly", slog.String("error", err.Error()))
logger.LogAttrs(ctx, slog.LevelError, "Failed to shutdown properly", slog.String("error", err.Error()))

}
log.Println("Shutdown trace provider")
logger.LogAttrs(ctx, slog.LevelInfo, "", slog.String("message", "Shotdown trace provider"))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
logger.LogAttrs(ctx, slog.LevelInfo, "", slog.String("message", "Shotdown trace provider"))
logger.LogAttrs(ctx, slog.LevelInfo, "Shutdown trace provider")

}()

var brokers string
mustMapEnv(&brokers, "KAFKA_SERVICE_ADDR")

brokerList := strings.Split(brokers, ",")
log.Printf("Kafka brokers: %s", strings.Join(brokerList, ", "))
logger.LogAttrs(ctx, slog.LevelInfo, "Kafka brokers", slog.String("Kafka brokers", strings.Join(brokerList, ",")))

ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM, syscall.SIGKILL)
defer cancel()
var consumerGroup sarama.ConsumerGroup
if consumerGroup, err = kafka.StartConsumerGroup(ctx, brokerList, log); err != nil {
log.Fatal(err)
if consumerGroup, err = kafka.StartConsumerGroup(ctx, brokerList, logger); err != nil {
logger.LogAttrs(ctx, slog.LevelError, "Failed to start consumer group", slog.String("error", err.Error()))
}
defer func() {
if err := consumerGroup.Close(); err != nil {
log.Printf("Error closing consumer group: %v", err)
logger.LogAttrs(ctx, slog.LevelError, "Error closing consumer group", slog.String("error", err.Error()))
}
log.Println("Closed consumer group")
logger.Log(ctx, slog.LevelInfo, "Closed consumer group")
}()

<-ctx.Done()

log.Println("Accounting service exited")
logger.Log(ctx, slog.LevelInfo, "message", "Accounting service exited")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
logger.Log(ctx, slog.LevelInfo, "message", "Accounting service exited")
logger.LogAttrs(ctx, slog.LevelInfo, "Accounting service exited")

}

func mustMapEnv(target *string, envKey string) {
Expand Down
Loading