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

GRPC Streaming add channel buffer per subscription and metrics #1687

Merged
merged 8 commits into from
Jun 12, 2024
Merged

GRPC Streaming add channel buffer per subscription and metrics #1687

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
20aaa96
fix metrics and add log on subscribe
jonfung-dydx Jun 11, 2024
76b5613
build and push jonfung
jonfung-dydx Jun 11, 2024
c77f0d5
additional channel per subscription
jonfung-dydx Jun 12, 2024
5da901d
immediately exit on error
jonfung-dydx Jun 12, 2024
eb8a11f
metrics on channel buffer lengths
jonfung-dydx Jun 12, 2024
ef874ab
more metrics
jonfung-dydx Jun 12, 2024
2516ca1
configurable cli for channel max length
jonfung-dydx Jun 12, 2024
191b79c
Revert "build and push jonfung"
jonfung-dydx Jun 12, 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
3 changes: 2 additions & 1 deletion protocol/app/app.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1958,7 +1958,8 @@ func getGrpcStreamingManagerFromOptions(
return streaming.NewGrpcStreamingManager(
logger,
appFlags.GrpcStreamingFlushIntervalMs,
appFlags.GrpcStreamingMaxBufferSize,
appFlags.GrpcStreamingMaxBatchSize,
appFlags.GrpcStreamingMaxChannelBufferSize,
Comment on lines +1961 to +1962
Copy link
Contributor

Choose a reason for hiding this comment

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

The GrpcStreamingManager constructor is called with incorrect parameters.

- return streaming.NewGrpcStreamingManager(
-     logger,
-     appFlags.GrpcStreamingFlushIntervalMs,
-     appFlags.GrpcStreamingMaxBatchSize,
-     appFlags.GrpcStreamingMaxChannelBufferSize,
- )
+ return streaming.NewGrpcStreamingManager(
+     logger,
+     time.Duration(appFlags.GrpcStreamingFlushIntervalMs) * time.Millisecond, // Ensure time duration is correctly formatted
+     appFlags.GrpcStreamingMaxBatchSize,
+     appFlags.GrpcStreamingMaxChannelBufferSize,
+ )
Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
appFlags.GrpcStreamingMaxBatchSize,
appFlags.GrpcStreamingMaxChannelBufferSize,
return streaming.NewGrpcStreamingManager(
logger,
time.Duration(appFlags.GrpcStreamingFlushIntervalMs) * time.Millisecond, // Ensure time duration is correctly formatted
appFlags.GrpcStreamingMaxBatchSize,
appFlags.GrpcStreamingMaxChannelBufferSize,
)

)
}
return streaming.NewNoopGrpcStreamingManager()
Expand Down
56 changes: 37 additions & 19 deletions protocol/app/flags/flags.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,10 @@ type Flags struct {
GrpcEnable bool

// Grpc Streaming
GrpcStreamingEnabled bool
GrpcStreamingFlushIntervalMs uint32
GrpcStreamingMaxBufferSize uint32
GrpcStreamingEnabled bool
GrpcStreamingFlushIntervalMs uint32
GrpcStreamingMaxBatchSize uint32
GrpcStreamingMaxChannelBufferSize uint32

VEOracleEnabled bool // Slinky Vote Extensions
}
Expand All @@ -40,9 +41,10 @@ const (
GrpcEnable = "grpc.enable"

// Grpc Streaming
GrpcStreamingEnabled = "grpc-streaming-enabled"
GrpcStreamingFlushIntervalMs = "grpc-streaming-flush-interval-ms"
GrpcStreamingMaxBufferSize = "grpc-streaming-max-buffer-size"
GrpcStreamingEnabled = "grpc-streaming-enabled"
GrpcStreamingFlushIntervalMs = "grpc-streaming-flush-interval-ms"
GrpcStreamingMaxBatchSize = "grpc-streaming-max-batch-size"
GrpcStreamingMaxChannelBufferSize = "grpc-streaming-max-channel-buffer-size"

// Slinky VEs enabled
VEOracleEnabled = "slinky-vote-extension-oracle-enabled"
Expand All @@ -55,9 +57,10 @@ const (
DefaultNonValidatingFullNode = false
DefaultDdErrorTrackingFormat = false

DefaultGrpcStreamingEnabled = false
DefaultGrpcStreamingFlushIntervalMs = 50
DefaultGrpcStreamingMaxBufferSize = 10000
DefaultGrpcStreamingEnabled = false
DefaultGrpcStreamingFlushIntervalMs = 50
DefaultGrpcStreamingMaxBatchSize = 10000
DefaultGrpcStreamingMaxChannelBufferSize = 10000

DefaultVEOracleEnabled = true
)
Expand Down Expand Up @@ -99,9 +102,14 @@ func AddFlagsToCmd(cmd *cobra.Command) {
"Flush interval (in ms) for grpc streaming",
)
cmd.Flags().Uint32(
GrpcStreamingMaxBufferSize,
DefaultGrpcStreamingMaxBufferSize,
"Maximum buffer size before grpc streaming cancels all connections",
GrpcStreamingMaxBatchSize,
DefaultGrpcStreamingMaxBatchSize,
"Maximum batch size before grpc streaming cancels all subscriptions",
)
cmd.Flags().Uint32(
GrpcStreamingMaxChannelBufferSize,
DefaultGrpcStreamingMaxChannelBufferSize,
"Maximum per-subscription channel size before grpc streaming cancels a singular subscription",
)
cmd.Flags().Bool(
VEOracleEnabled,
Expand All @@ -122,12 +130,15 @@ func (f *Flags) Validate() error {
if !f.GrpcEnable {
return fmt.Errorf("grpc.enable must be set to true - grpc streaming requires gRPC server")
}
if f.GrpcStreamingMaxBufferSize == 0 {
return fmt.Errorf("grpc streaming buffer size must be positive number")
if f.GrpcStreamingMaxBatchSize == 0 {
return fmt.Errorf("grpc streaming batch size must be positive number")
}
if f.GrpcStreamingFlushIntervalMs == 0 {
return fmt.Errorf("grpc streaming flush interval must be positive number")
}
if f.GrpcStreamingMaxChannelBufferSize == 0 {
return fmt.Errorf("grpc streaming channel size must be positive number")
}
}
return nil
}
Expand All @@ -148,9 +159,10 @@ func GetFlagValuesFromOptions(
GrpcAddress: config.DefaultGRPCAddress,
GrpcEnable: true,

GrpcStreamingEnabled: DefaultGrpcStreamingEnabled,
GrpcStreamingFlushIntervalMs: DefaultGrpcStreamingFlushIntervalMs,
GrpcStreamingMaxBufferSize: DefaultGrpcStreamingMaxBufferSize,
GrpcStreamingEnabled: DefaultGrpcStreamingEnabled,
GrpcStreamingFlushIntervalMs: DefaultGrpcStreamingFlushIntervalMs,
GrpcStreamingMaxBatchSize: DefaultGrpcStreamingMaxBatchSize,
GrpcStreamingMaxChannelBufferSize: DefaultGrpcStreamingMaxChannelBufferSize,

VEOracleEnabled: true,
}
Expand Down Expand Up @@ -204,9 +216,15 @@ func GetFlagValuesFromOptions(
}
}

if option := appOpts.Get(GrpcStreamingMaxBufferSize); option != nil {
if option := appOpts.Get(GrpcStreamingMaxBatchSize); option != nil {
if v, err := cast.ToUint32E(option); err == nil {
result.GrpcStreamingMaxBatchSize = v
}
}

if option := appOpts.Get(GrpcStreamingMaxChannelBufferSize); option != nil {
if v, err := cast.ToUint32E(option); err == nil {
result.GrpcStreamingMaxBufferSize = v
result.GrpcStreamingMaxChannelBufferSize = v
}
}

Expand Down
114 changes: 69 additions & 45 deletions protocol/app/flags/flags_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,11 @@ func TestAddFlagsToCommand(t *testing.T) {
fmt.Sprintf("Has %s flag", flags.GrpcStreamingFlushIntervalMs): {
flagName: flags.GrpcStreamingFlushIntervalMs,
},
fmt.Sprintf("Has %s flag", flags.GrpcStreamingMaxBufferSize): {
flagName: flags.GrpcStreamingMaxBufferSize,
fmt.Sprintf("Has %s flag", flags.GrpcStreamingMaxBatchSize): {
flagName: flags.GrpcStreamingMaxBatchSize,
},
fmt.Sprintf("Has %s flag", flags.GrpcStreamingMaxChannelBufferSize): {
flagName: flags.GrpcStreamingMaxChannelBufferSize,
},
}

Expand Down Expand Up @@ -69,11 +72,12 @@ func TestValidate(t *testing.T) {
},
"success - gRPC streaming enabled for validating nodes": {
flags: flags.Flags{
NonValidatingFullNode: false,
GrpcEnable: true,
GrpcStreamingEnabled: true,
GrpcStreamingFlushIntervalMs: 100,
GrpcStreamingMaxBufferSize: 10000,
NonValidatingFullNode: false,
GrpcEnable: true,
GrpcStreamingEnabled: true,
GrpcStreamingFlushIntervalMs: 100,
GrpcStreamingMaxBatchSize: 10000,
GrpcStreamingMaxChannelBufferSize: 10000,
},
},
"failure - gRPC disabled": {
Expand All @@ -90,26 +94,37 @@ func TestValidate(t *testing.T) {
},
expectedErr: fmt.Errorf("grpc.enable must be set to true - grpc streaming requires gRPC server"),
},
"failure - gRPC streaming enabled with zero buffer size": {
"failure - gRPC streaming enabled with zero batch size": {
flags: flags.Flags{
NonValidatingFullNode: true,
GrpcEnable: true,
GrpcStreamingEnabled: true,
GrpcStreamingFlushIntervalMs: 100,
GrpcStreamingMaxBufferSize: 0,
GrpcStreamingMaxBatchSize: 0,
},
expectedErr: fmt.Errorf("grpc streaming buffer size must be positive number"),
expectedErr: fmt.Errorf("grpc streaming batch size must be positive number"),
},
"failure - gRPC streaming enabled with zero flush interval ms": {
flags: flags.Flags{
NonValidatingFullNode: true,
GrpcEnable: true,
GrpcStreamingEnabled: true,
GrpcStreamingFlushIntervalMs: 0,
GrpcStreamingMaxBufferSize: 10000,
GrpcStreamingMaxBatchSize: 10000,
},
expectedErr: fmt.Errorf("grpc streaming flush interval must be positive number"),
},
"failure - gRPC streaming enabled with zero channel size ms": {
flags: flags.Flags{
NonValidatingFullNode: true,
GrpcEnable: true,
GrpcStreamingEnabled: true,
GrpcStreamingFlushIntervalMs: 100,
GrpcStreamingMaxBatchSize: 10000,
GrpcStreamingMaxChannelBufferSize: 0,
},
expectedErr: fmt.Errorf("grpc streaming channel size must be positive number"),
},
}
for name, tc := range tests {
t.Run(name, func(t *testing.T) {
Expand All @@ -129,44 +144,48 @@ func TestGetFlagValuesFromOptions(t *testing.T) {
optsMap map[string]any

// Expectations.
expectedNonValidatingFullNodeFlag bool
expectedDdAgentHost string
expectedDdTraceAgentPort uint16
expectedGrpcAddress string
expectedGrpcEnable bool
expectedGrpcStreamingEnable bool
expectedGrpcStreamingFlushMs uint32
expectedGrpcStreamingBufferSize uint32
expectedNonValidatingFullNodeFlag bool
expectedDdAgentHost string
expectedDdTraceAgentPort uint16
expectedGrpcAddress string
expectedGrpcEnable bool
expectedGrpcStreamingEnable bool
expectedGrpcStreamingFlushMs uint32
expectedGrpcStreamingBatchSize uint32
expectedGrpcStreamingMaxChannelBufferSize uint32
}{
"Sets to default if unset": {
expectedNonValidatingFullNodeFlag: false,
expectedDdAgentHost: "",
expectedDdTraceAgentPort: 8126,
expectedGrpcAddress: "localhost:9090",
expectedGrpcEnable: true,
expectedGrpcStreamingEnable: false,
expectedGrpcStreamingFlushMs: 50,
expectedGrpcStreamingBufferSize: 10000,
expectedNonValidatingFullNodeFlag: false,
expectedDdAgentHost: "",
expectedDdTraceAgentPort: 8126,
expectedGrpcAddress: "localhost:9090",
expectedGrpcEnable: true,
expectedGrpcStreamingEnable: false,
expectedGrpcStreamingFlushMs: 50,
expectedGrpcStreamingBatchSize: 10000,
expectedGrpcStreamingMaxChannelBufferSize: 10000,
},
"Sets values from options": {
optsMap: map[string]any{
flags.NonValidatingFullNodeFlag: true,
flags.DdAgentHost: "agentHostTest",
flags.DdTraceAgentPort: uint16(777),
flags.GrpcEnable: false,
flags.GrpcAddress: "localhost:9091",
flags.GrpcStreamingEnabled: "true",
flags.GrpcStreamingFlushIntervalMs: uint32(408),
flags.GrpcStreamingMaxBufferSize: uint32(650),
flags.NonValidatingFullNodeFlag: true,
flags.DdAgentHost: "agentHostTest",
flags.DdTraceAgentPort: uint16(777),
flags.GrpcEnable: false,
flags.GrpcAddress: "localhost:9091",
flags.GrpcStreamingEnabled: "true",
flags.GrpcStreamingFlushIntervalMs: uint32(408),
flags.GrpcStreamingMaxBatchSize: uint32(650),
flags.GrpcStreamingMaxChannelBufferSize: uint32(972),
},
expectedNonValidatingFullNodeFlag: true,
expectedDdAgentHost: "agentHostTest",
expectedDdTraceAgentPort: 777,
expectedGrpcEnable: false,
expectedGrpcAddress: "localhost:9091",
expectedGrpcStreamingEnable: true,
expectedGrpcStreamingFlushMs: 408,
expectedGrpcStreamingBufferSize: 650,
expectedNonValidatingFullNodeFlag: true,
expectedDdAgentHost: "agentHostTest",
expectedDdTraceAgentPort: 777,
expectedGrpcEnable: false,
expectedGrpcAddress: "localhost:9091",
expectedGrpcStreamingEnable: true,
expectedGrpcStreamingFlushMs: 408,
expectedGrpcStreamingBatchSize: 650,
expectedGrpcStreamingMaxChannelBufferSize: 972,
},
}

Expand Down Expand Up @@ -216,8 +235,13 @@ func TestGetFlagValuesFromOptions(t *testing.T) {
)
require.Equal(
t,
tc.expectedGrpcStreamingBufferSize,
flags.GrpcStreamingMaxBufferSize,
tc.expectedGrpcStreamingBatchSize,
flags.GrpcStreamingMaxBatchSize,
)
require.Equal(
t,
tc.expectedGrpcStreamingMaxChannelBufferSize,
flags.GrpcStreamingMaxChannelBufferSize,
)
})
}
Expand Down
4 changes: 3 additions & 1 deletion protocol/lib/metrics/metric_keys.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,11 +69,13 @@ const (
GrpcSendOrderbookUpdatesLatency = "grpc_send_orderbook_updates_latency"
GrpcSendOrderbookSnapshotLatency = "grpc_send_orderbook_snapshot_latency"
GrpcSendOrderbookFillsLatency = "grpc_send_orderbook_fills_latency"
GrpcEmitProtocolUpdateCount = "grpc_emit_protocol_update_count"
GrpcAddUpdateToBufferCount = "grpc_add_update_to_buffer_count"
GrpcAddToSubscriptionChannelCount = "grpc_add_to_subscription_channel_count"
GrpcSendResponseToSubscriberCount = "grpc_send_response_to_subscriber_count"
GrpcStreamSubscriberCount = "grpc_stream_subscriber_count"
GrpcStreamNumUpdatesBuffered = "grpc_stream_num_updates_buffered"
GrpcFlushUpdatesLatency = "grpc_flush_updates_latency"
GrpcSubscriptionChannelLength = "grpc_subscription_channel_length"

EndBlocker = "end_blocker"
EndBlockerLag = "end_blocker_lag"
Expand Down
Loading
Loading