Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions deploy/featureflags/local.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,4 @@ flags:
V3MirrorFraction: 1.0
UseSpannerGraph: true
UseStaleReads: true
V2DivertFraction: 1.0
9 changes: 9 additions & 0 deletions internal/featureflags/featureflags.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ type Flags struct {
UseStaleReads bool `yaml:"UseStaleReads"`
// Whether to enable the embeddings resolver.
EnableEmbeddingsResolver bool `yaml:"EnableEmbeddingsResolver"`
// Fraction of V2 API requests to divert to the new dispatcher backend. Value from 0 to 1.0.
V2DivertFraction float64 `yaml:"V2DivertFraction"`
}

// setDefaultValues creates a new Flags struct with default values.
Expand All @@ -49,6 +51,7 @@ func setDefaultValues() *Flags {
SpannerGraphDatabase: "",
UseStaleReads: false,
EnableEmbeddingsResolver: false,
V2DivertFraction: 0.0,
}
}

Expand All @@ -66,6 +69,12 @@ func (f *Flags) validateFlagValues() error {
if f.UseStaleReads && !f.UseSpannerGraph {
return fmt.Errorf("UseStaleReads requires UseSpannerGraph to be true")
}
if f.V2DivertFraction < 0 || f.V2DivertFraction > 1.0 {
return fmt.Errorf("V2DivertFraction must be between 0 and 1.0, got %f", f.V2DivertFraction)
}
if f.V2DivertFraction > 0 && !f.EnableV3 {
return fmt.Errorf("V2DivertFraction > 0 requires EnableV3 to be true")
}
return nil
}

Expand Down
12 changes: 12 additions & 0 deletions internal/server/handler_v2.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,14 @@ package server
import (
"context"
"log/slog"
"math/rand"
"time"

"github.com/datacommonsorg/mixer/internal/log"
"github.com/datacommonsorg/mixer/internal/merger"
pb "github.com/datacommonsorg/mixer/internal/proto"
pbv2 "github.com/datacommonsorg/mixer/internal/proto/v2"
"github.com/datacommonsorg/mixer/internal/server/datasources"
"github.com/datacommonsorg/mixer/internal/server/pagination"
"github.com/datacommonsorg/mixer/internal/server/statvar/search"
"github.com/datacommonsorg/mixer/internal/server/translator"
Expand Down Expand Up @@ -105,6 +107,11 @@ func (s *Server) V2Resolve(
func (s *Server) V2Node(ctx context.Context, in *pbv2.NodeRequest) (
*pbv2.NodeResponse, error,
) {
if rand.Float64() < s.flags.V2DivertFraction {
slog.Info("V2Node request diverted to dispatcher backend", "request", in)
return s.dispatcher.Node(ctx, in, datasources.DefaultPageSize)
Copy link
Contributor

Choose a reason for hiding this comment

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

nit: Should we add a log message when traffic is diverted? If we do, add for obs as well.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Good idea - aded a basic slog message for now. As a follow up I can try to add more detailed metrics for tracking this

}

v2StartTime := time.Now()
errGroup, errCtx := errgroup.WithContext(ctx)
localRespChan := make(chan *pbv2.NodeResponse, 1)
Expand Down Expand Up @@ -260,6 +267,11 @@ func (s *Server) V2Event(
func (s *Server) V2Observation(
ctx context.Context, in *pbv2.ObservationRequest,
) (*pbv2.ObservationResponse, error) {
if rand.Float64() < s.flags.V2DivertFraction {
slog.Info("V2Observation request diverted to dispatcher backend", "request", in)
return s.dispatcher.Observation(ctx, in)
}

v2StartTime := time.Now()

surface, toRemote := util.GetMetadata(ctx)
Expand Down
Loading