converted some spans

Author: darkcoderrises

conn/node.go

@@ -21,7 +21,8 @@ import (
 	"github.com/pkg/errors"
 	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/raft/v3/raftpb"
-	otrace "go.opencensus.io/trace"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/dgraph-io/badger/v4/y"
@@ -432,7 +433,7 @@ func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
 	}
 
 	c := pb.NewRaftClient(pool.Get())
-	ctx, span := otrace.StartSpan(context.Background(),
+	ctx, span := otel.Tracer("").Start(context.Background(),
 		fmt.Sprintf("RaftMessage-%d-to-%d", n.Id, to))
 	defer span.End()
 
@@ -473,15 +474,15 @@ func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
 				Payload: &api.Payload{Data: data},
 			}
 			slurp(batch) // Pick up more entries from msgCh, if present.
-			span.Annotatef(nil, "[to: %x] [Packets: %d] Sending data of length: %d.",
-				to, packets, len(batch.Payload.Data))
+			span.AddEvent(fmt.Sprintf("[to: %x] [Packets: %d] Sending data of length: %d.",
+				to, packets, len(batch.Payload.Data)))
 			if packets%10000 == 0 {
 				glog.V(2).Infof("[to: %x] [Packets: %d] Sending data of length: %d.",
 					to, packets, len(batch.Payload.Data))
 			}
 			packets++
 			if err := mc.Send(batch); err != nil {
-				span.Annotatef(nil, "Error while mc.Send: %v", err)
+				span.AddEvent(fmt.Sprintf("Error while mc.Send: %v", err))
 				glog.Errorf("[to: %x] Error while mc.Send: %v", to, err)
 				switch {
 				case strings.Contains(err.Error(), "TransientFailure"):
@@ -513,8 +514,7 @@ func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
 			}
 		case <-ticker.C:
 			if lastPackets == packets {
-				span.Annotatef(nil,
-					"No activity for a while [Packets == %d]. Closing connection.", packets)
+				span.AddEvent(fmt.Sprintf("No activity for a while [Packets == %d]. Closing connection.", packets))
 				return mc.CloseSend()
 			}
 			lastPackets = packets
@@ -640,34 +640,34 @@ var readIndexOk, readIndexTotal uint64
 
 // WaitLinearizableRead waits until a linearizable read can be performed.
 func (n *Node) WaitLinearizableRead(ctx context.Context) error {
-	span := otrace.FromContext(ctx)
-	span.Annotate(nil, "WaitLinearizableRead")
+	span := trace.SpanFromContext(ctx)
+	span.AddEvent("WaitLinearizableRead")
 
 	if num := atomic.AddUint64(&readIndexTotal, 1); num%1000 == 0 {
 		glog.V(2).Infof("ReadIndex Total: %d\n", num)
 	}
 	indexCh := make(chan uint64, 1)
 	select {
 	case n.requestCh <- linReadReq{indexCh: indexCh}:
-		span.Annotate(nil, "Pushed to requestCh")
+		span.AddEvent("Pushed to requestCh")
 	case <-ctx.Done():
-		span.Annotate(nil, "Context expired")
+		span.AddEvent("Context expired")
 		return ctx.Err()
 	}
 
 	select {
 	case index := <-indexCh:
-		span.Annotatef(nil, "Received index: %d", index)
+		span.AddEvent(fmt.Sprintf("Received index %d", index))
 		if index == 0 {
 			return errReadIndex
 		} else if num := atomic.AddUint64(&readIndexOk, 1); num%1000 == 0 {
 			glog.V(2).Infof("ReadIndex OK: %d\n", num)
 		}
 		err := n.Applied.WaitForMark(ctx, index)
-		span.Annotatef(nil, "Error from Applied.WaitForMark: %v", err)
+		span.AddEvent(fmt.Sprintf("Error from Applied.WaitForMark: %v", err))
 		return err
 	case <-ctx.Done():
-		span.Annotate(nil, "Context expired")
+		span.AddEvent("Context expired")
 		return ctx.Err()
 	}
 }

conn/pool.go

@@ -14,7 +14,7 @@ import (
 
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
-	"go.opencensus.io/plugin/ocgrpc"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -159,7 +159,7 @@ func (p *Pools) Connect(addr string, tlsClientConf *tls.Config) *Pool {
 // newPool creates a new "pool" with one gRPC connection, refcount 0.
 func newPool(addr string, tlsClientConf *tls.Config) (*Pool, error) {
 	conOpts := []grpc.DialOption{
-		grpc.WithStatsHandler(&ocgrpc.ClientHandler{}),
+		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallRecvMsgSize(x.GrpcMaxSize),
 			grpc.MaxCallSendMsgSize(x.GrpcMaxSize),

conn/raft_server.go

@@ -8,6 +8,7 @@ package conn
 import (
 	"context"
 	"encoding/binary"
+	"fmt"
 	"math/rand"
 	"strconv"
 	"sync"
@@ -17,7 +18,7 @@ import (
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
 	"go.etcd.io/etcd/raft/v3/raftpb"
-	otrace "go.opencensus.io/trace"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/dgraph-io/dgo/v240/protos/api"
 	"github.com/hypermodeinc/dgraph/v24/protos/pb"
@@ -183,13 +184,13 @@ func (w *RaftServer) RaftMessage(server pb.Raft_RaftMessageServer) error {
 	if ctx.Err() != nil {
 		return ctx.Err()
 	}
-	span := otrace.FromContext(ctx)
+	span := trace.SpanFromContext(ctx)
 
 	node := w.GetNode()
 	if node == nil || node.Raft() == nil {
 		return ErrNoNode
 	}
-	span.Annotatef(nil, "Stream server is node %#x", node.Id)
+	span.AddEvent(fmt.Sprintf("Stream server is node %d", node.Id))
 
 	var rc *pb.RaftContext
 	raft := node.Raft()
@@ -247,7 +248,7 @@ func (w *RaftServer) RaftMessage(server pb.Raft_RaftMessageServer) error {
 		}
 		if loop == 1 {
 			rc = batch.GetContext()
-			span.Annotatef(nil, "Stream from %#x", rc.GetId())
+			span.AddEvent(fmt.Sprintf("Stream from %#x", rc.GetId()))
 			if rc != nil {
 				node.Connect(rc.Id, rc.Addr)
 			}

dgraph/cmd/alpha/run.go

@@ -28,8 +28,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 	"go.opencensus.io/plugin/ocgrpc"
-	otrace "go.opencensus.io/trace"
-	"go.opencensus.io/zpages"
+	"go.opentelemetry.io/contrib/zpages"
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -509,7 +508,7 @@ func setupServer(closer *z.Closer) {
 	baseMux.HandleFunc("/health", healthCheck)
 	baseMux.HandleFunc("/state", stateHandler)
 	baseMux.HandleFunc("/debug/jemalloc", x.JemallocHandler)
-	zpages.Handle(baseMux, "/debug/z")
+	http.DefaultServeMux.Handle("/debug/z", zpages.NewTracezHandler(zpages.NewSpanProcessor()))
 
 	// TODO: Figure out what this is for?
 	http.HandleFunc("/debug/store", storeStatsHandler)
@@ -770,10 +769,6 @@ func run() {
 			return true, true
 		}
 	}
-	otrace.ApplyConfig(otrace.Config{
-		DefaultSampler:             otrace.ProbabilitySampler(x.WorkerConfig.Trace.GetFloat64("ratio")),
-		MaxAnnotationEventsPerSpan: 256,
-	})
 
 	// Posting will initialize index which requires schema. Hence, initialize
 	// schema before calling posting.Init().

dgraph/cmd/zero/run.go

@@ -20,9 +20,9 @@ import (
 
 	"github.com/golang/glog"
 	"github.com/spf13/cobra"
-	"go.opencensus.io/plugin/ocgrpc"
 	otrace "go.opencensus.io/trace"
-	"go.opencensus.io/zpages"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/contrib/zpages"
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -145,7 +145,7 @@ func (st *state) serveGRPC(l net.Listener, store *raftwal.DiskStorage) {
 		grpc.MaxRecvMsgSize(x.GrpcMaxSize),
 		grpc.MaxSendMsgSize(x.GrpcMaxSize),
 		grpc.MaxConcurrentStreams(1000),
-		grpc.StatsHandler(&ocgrpc.ServerHandler{}),
+		grpc.StatsHandler(otelgrpc.NewClientHandler()),
 		grpc.UnaryInterceptor(audit.AuditRequestGRPC),
 	}
 
@@ -314,7 +314,7 @@ func run() {
 		baseMux.HandleFunc("/enterpriseLicense", st.applyEnterpriseLicense)
 	}
 	baseMux.HandleFunc("/debug/jemalloc", x.JemallocHandler)
-	zpages.Handle(baseMux, "/debug/z")
+	http.DefaultServeMux.Handle("/debug/z", zpages.NewTracezHandler(zpages.NewSpanProcessor()))
 
 	// This must be here. It does not work if placed before Grpc init.
 	x.Check(st.node.initAndStartNode())

edgraph/access_ee.go

@@ -18,7 +18,9 @@ import (
 	"github.com/golang-jwt/jwt/v5"
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
-	otrace "go.opencensus.io/trace"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
@@ -57,7 +59,7 @@ func (s *Server) Login(ctx context.Context,
 			"supplying the appropriate license file to Dgraph Zero using the HTTP endpoint.")
 	}
 
-	ctx, span := otrace.StartSpan(ctx, "server.Login")
+	ctx, span := otel.Tracer("").Start(ctx, "server.Login")
 	defer span.End()
 
 	// record the client ip for this login request
@@ -66,9 +68,7 @@ func (s *Server) Login(ctx context.Context,
 		return nil, err
 	} else {
 		addr = ipAddr.String()
-		span.Annotate([]otrace.Attribute{
-			otrace.StringAttribute("client_ip", addr),
-		}, "client ip for login")
+		span.AddEvent("client ip for login", trace.WithAttributes(attribute.String("client_ip", addr)))
 	}
 
 	user, err := s.authenticateLogin(ctx, request)
@@ -730,16 +730,14 @@ func authorizeAlter(ctx context.Context, op *api.Operation) error {
 	}
 
 	err := doAuthorizeAlter()
-	span := otrace.FromContext(ctx)
-	if span != nil {
-		span.Annotatef(nil, (&accessEntry{
-			userId:    userId,
-			groups:    groupIds,
-			preds:     preds,
-			operation: acl.Modify,
-			allowed:   err == nil,
-		}).String())
-	}
+	span := trace.SpanFromContext(ctx)
+	span.AddEvent((&accessEntry{
+		userId:    userId,
+		groups:    groupIds,
+		preds:     preds,
+		operation: acl.Modify,
+		allowed:   err == nil,
+	}).String())
 
 	return err
 }
@@ -851,9 +849,9 @@ func authorizeMutation(ctx context.Context, gmu *dql.Mutation) error {
 
 	err := doAuthorizeMutation()
 
-	span := otrace.FromContext(ctx)
+	span := trace.SpanFromContext(ctx)
 	if span != nil {
-		span.Annotatef(nil, (&accessEntry{
+		span.AddEvent((&accessEntry{
 			userId:    userId,
 			groups:    groupIds,
 			preds:     preds,
@@ -1008,8 +1006,8 @@ func authorizeQuery(ctx context.Context, parsedReq *dql.Result, graphql bool) er
 		return err
 	}
 
-	if span := otrace.FromContext(ctx); span != nil {
-		span.Annotatef(nil, (&accessEntry{
+	if span := trace.SpanFromContext(ctx); span != nil {
+		span.AddEvent((&accessEntry{
 			userId:    userId,
 			groups:    groupIds,
 			preds:     preds,

edgraph/alter.go

@@ -18,10 +18,12 @@ import (
 	"github.com/hypermodeinc/dgraph/v24/schema"
 	"github.com/hypermodeinc/dgraph/v24/worker"
 	"github.com/hypermodeinc/dgraph/v24/x"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
-	otrace "go.opencensus.io/trace"
 	"google.golang.org/grpc/status"
 )
 
@@ -249,9 +251,9 @@ func executeSetSchema(ctx context.Context, startTs uint64, req *apiv25.AlterRequ
 
 // Alter handles requests to change the schema or remove parts or all of the data.
 func (s *ServerV25) Alter(ctx context.Context, req *apiv25.AlterRequest) (*apiv25.AlterResponse, error) {
-	ctx, span := otrace.StartSpan(ctx, "ServerV25.Alter")
+	ctx, span := otel.Tracer("").Start(ctx, "ServerV25.Alter")
 	defer span.End()
-	span.Annotatef(nil, "Alter operation: %+v", req)
+	span.AddEvent("Alter operation", trace.WithAttributes(attribute.String("request", req.String())))
 
 	// Always print out Alter operations because they are important and rare.
 	glog.Infof("Received ALTER op: %+v", req)