Add a logging observability service implementation in the lib

[presmihaylov]

Currently, all tutorials prompt you to init cloudevents client like this:

c, err := cloudevents.NewClient(protocol)

What they're not telling you is that all event errors due to eg malformed event or etc will silently fail instead of at least logging that something is wrong.

To capture those and at least log them, you need to use:

c, err := cloudevents.NewClient(protocol, client.WithObservabilityService(&yourObservabilitySvc{}))

but it's tedious that:

• Default behavior is to silently fail. I was debugging for a whole day to find rootcause of issue. Resorted to debugging library code in the end to figure out what's wrong.
• The library doesn't provide you with a sane observability service implementation that logs event errors in a sensible way

My proposal is to 1) add a LoggingObservabilityService implementation within the library somewhere and 2) provide that as the default vs. the noOpObservabilityService which is the default atm.

[embano1]

Just to double-check, you are not talking about the logging capabilities we already have (see screenshot)? Can you give an example of an error path where you see the behavior you described?

[presmihaylov]

Yes, I am not referring to that (although it's useful to be aware of that, hadn't seen that).

Here's how to reproduce the issue.

Receiver:

package main

import (
	"context"
	"fmt"
	"log"

	cepubsub "github.com/cloudevents/sdk-go/protocol/pubsub/v2"
	cloudevents "github.com/cloudevents/sdk-go/v2"
	"github.com/cloudevents/sdk-go/v2/binding"
	"github.com/cloudevents/sdk-go/v2/client"
	"github.com/cloudevents/sdk-go/v2/event"
)

func receive(ctx context.Context, event event.Event) error {
	fmt.Printf("------------------ Received MSG ------------------\n")
	return nil
}

func main() {
	ctx := context.Background()
	t, err := cepubsub.New(context.Background(),
		cepubsub.WithProjectID("helloworld"),
		cepubsub.AllowCreateTopic(true),
		cepubsub.AllowCreateSubscription(true),
		cepubsub.WithSubscriptionAndTopicID("helloworld", "helloworld"))
	if err != nil {
		log.Fatalf("failed to create pubsub protocol, %s", err.Error())
	}
	c, err := cloudevents.NewClient(t)

	if err != nil {
		log.Fatalf("failed to create client, %s", err.Error())
	}

	log.Println("Created client, listening...")

	if err := c.StartReceiver(ctx, receive); err != nil {
		log.Fatalf("failed to start pubsub receiver, %s", err.Error())
	}
}

Send a message using this python publisher mentioned in official docs:

python3 publisher.py helloworld publish helloworld

The result is that the message produced by this python publisher is malformed (no clue why, but it is). However, the receiver gives 0 indication that this is the case. It just appears as if it's not consuming anything.

However, if you add this:

	c, err := cloudevents.NewClient(t, client.WithObservabilityService(&myObservabilityService{}))

...

type myObservabilityService struct{}

func (n myObservabilityService) InboundContextDecorators() []func(context.Context, binding.Message) context.Context {
	return nil
}

func (n myObservabilityService) RecordReceivedMalformedEvent(ctx context.Context, err error) {
	fmt.Println("ERROR", err)
}

func (n myObservabilityService) RecordCallingInvoker(ctx context.Context, event *event.Event) (context.Context, func(errOrResult error)) {
	return ctx, func(errOrResult error) {}
}

func (n myObservabilityService) RecordSendingEvent(ctx context.Context, event event.Event) (context.Context, func(errOrResult error)) {
	return ctx, func(errOrResult error) {}
}

func (n myObservabilityService) RecordRequestEvent(ctx context.Context, e event.Event) (context.Context, func(errOrResult error, event *event.Event)) {
	return ctx, func(errOrResult error, event *event.Event) {}
}

You're starting to get the malformed message error:

$ go run *.go
2023/02/11 14:10:12 Created client, listening...
{"level":"info","ts":1676117412.388352,"logger":"fallback","caller":"v2@v2.13.0/protocol.go:195","msg":"starting subscriber for Topic \"helloworld\", Subscription \"helloworld\""}
{"level":"info","ts":1676117412.388633,"logger":"fallback","caller":"v2@v2.13.0/protocol.go:198","msg":"conn is&{true true helloworld 0x14000380380 helloworld <nil> helloworld <nil> {0 0} <nil> <nil> <nil> false }"}
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
ERROR unknown Message encoding
...

[embano1]

Can you do me a favor and create a zap Development logger and inject it into the initial ctx so CE uses this logger? I see this line in client.Request which could surface this error. Just trying to see if there's an easy fix here:

sdk-go/v2/client/client.go

Lines 178 to 184 in 66eacad

	// try to turn msg into an event, it might not work and that is ok.
	if rs, rserr := binding.ToEvent(ctx, msg); rserr != nil {
	cecontext.LoggerFrom(ctx).Debugw("response: failed calling ToEvent", zap.Error(rserr), zap.Any("resp", msg))
	// If the protocol returns no error, it is an ACK on the request, but we had
	// issues turning the response into an event, so make an ACK Result and pass
	// down the ToEvent error as well.
	err = protocol.NewReceipt(true, "failed to convert response into event: %v\n%w", rserr, err)

[presmihaylov]

didn't work unfortunately. Here's my updated receiver \w logger plugged in:

package main

import (
	"context"
	"fmt"
	"log"

	cepubsub "github.com/cloudevents/sdk-go/protocol/pubsub/v2"
	cloudevents "github.com/cloudevents/sdk-go/v2"
	cecontext "github.com/cloudevents/sdk-go/v2/context"
	"github.com/cloudevents/sdk-go/v2/event"
	"go.uber.org/zap"
)

func receive(ctx context.Context, event event.Event) error {
	fmt.Printf("------------------ Received MSG ------------------\n")
	return nil
}

func main() {
	ctx := context.Background()
	logger, err := zap.NewDevelopment()
	if err != nil {
		panic(err)
	}
	defer logger.Sync()

	ctx = cecontext.WithLogger(ctx, logger.Sugar())
	t, err := cepubsub.New(ctx,
		cepubsub.WithProjectID("helloworld"),
		cepubsub.AllowCreateTopic(true),
		cepubsub.AllowCreateSubscription(true),
		cepubsub.WithSubscriptionAndTopicID("helloworld", "helloworld"))
	if err != nil {
		log.Fatalf("failed to create pubsub protocol, %s", err.Error())
	}
	c, err := cloudevents.NewClient(t)

	if err != nil {
		log.Fatalf("failed to create client, %s", err.Error())
	}

	log.Println("Created client, listening...")

	if err := c.StartReceiver(ctx, receive); err != nil {
		log.Fatalf("failed to start pubsub receiver, %s", err.Error())
	}
}