Mirror of https://github.com/roostorg/osprey
github.com/roostorg/osprey
Fix osprey coordinator compilation (#57)
+186
-171
+4
-5
osprey_coordinator/Cargo.toml
+4
-5
osprey_coordinator/Cargo.toml
···
1
1
[package]
2
-
name = "smite_coordinator"
2
+
name = "osprey_coordinator"
3
3
version = "0.1.0"
4
4
edition = "2021"
5
5
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
···
20
20
dynamic-pool = "0.2.2"
21
21
env_logger = "0.7"
22
22
etcd = { git = "https://github.com/discord/rust-etcd", rev = "e3fc8242de7303582feca89d90230e89e744d06e" }
23
-
etcd_config_derive = { path = "etcd_config_derive", package = "smite_coordinator_etcd_config_derive" }
23
+
etcd_config_derive = { path = "etcd_config_derive", package = "osprey_coordinator_etcd_config_derive" }
24
24
futures = "0.3"
25
25
fxhash = "0.2"
26
26
goauth = "0.10.0"
···
35
35
lazy_static = "1.4.0"
36
36
log = "0.4"
37
37
md5 = "0.7.0"
38
-
metrics_derive = { path = "metrics_derive", package = "smite_coordinator_metrics_derive" }
38
+
metrics_derive = { path = "metrics_derive", package = "osprey_coordinator_metrics_derive" }
39
39
msgpack_simple = "1.0.2"
40
40
num_cpus = "1.0"
41
41
opentelemetry = "0.23"
···
56
56
serde = { version = "1.0", features = ["rc", "derive"] }
57
57
serde-transcode = "1.1.1"
58
58
serde_json = "1.0"
59
-
smite_data_services_common = { path = "../common" }
60
59
smpl_jwt = "0.6.0"
61
60
strum_macros = "0.24"
62
61
thiserror = "2.0.12"
···
87
86
[dev-dependencies]
88
87
89
88
[[bin]]
90
-
name = "smite_coordinator"
89
+
name = "osprey_coordinator"
91
90
path = "src/main.rs"
+22
-22
osprey_coordinator/build.rs
+22
-22
osprey_coordinator/build.rs
···
1
1
fn main() -> Result<(), Box<dyn std::error::Error>> {
2
-
let proto_root = std::path::Path::new("../../proto");
3
-
let includes = vec!["../../proto/"];
2
+
let proto_root = std::path::Path::new("../proto");
3
+
let includes = vec!["../proto/"];
4
4
5
5
let proto_paths = [
6
-
proto_root.join("proto/osprey/rpc/actions/v1/action.proto"),
7
-
proto_root.join("proto/osprey/rpc/smite_coordinator/sync_action/v1/service.proto"),
6
+
proto_root.join("osprey/rpc/actions/v1/action.proto"),
7
+
proto_root.join("osprey/rpc/osprey_coordinator/sync_action/v1/service.proto"),
8
8
proto_root
9
-
.join("proto/osprey/rpc/smite_coordinator/bidirectional_stream/v1/service.proto"),
10
-
proto_root.join("proto/osprey/rpc/common/v1/*.proto"),
9
+
.join("osprey/rpc/osprey_coordinator/bidirectional_stream/v1/service.proto"),
10
+
proto_root.join("osprey/rpc/common/v1/*.proto"),
11
11
]
12
12
.into_iter()
13
13
.map(|pattern| glob::glob(pattern.to_str().unwrap()))
···
37
37
// Compile Google Cloud protos for inlined gcloud module
38
38
tonic_build::configure().build_server(false).compile(
39
39
&[
40
-
proto_root.join("proto/google/api/annotations.proto"),
41
-
proto_root.join("proto/google/api/client.proto"),
42
-
proto_root.join("proto/google/api/field_behavior.proto"),
43
-
proto_root.join("proto/google/api/http.proto"),
44
-
proto_root.join("proto/google/api/resource.proto"),
45
-
proto_root.join("proto/google/iam/v1/iam_policy.proto"),
46
-
proto_root.join("proto/google/iam/v1/options.proto"),
47
-
proto_root.join("proto/google/iam/v1/policy.proto"),
48
-
proto_root.join("proto/google/longrunning/operations.proto"),
49
-
proto_root.join("proto/google/rpc/status.proto"),
50
-
proto_root.join("proto/google/type/expr.proto"),
51
-
proto_root.join("proto/google/pubsub/v1/pubsub.proto"),
52
-
proto_root.join("proto/google/cloud/kms/v1/service.proto"),
53
-
proto_root.join("proto/google/cloud/kms/v1/resources.proto"),
54
-
proto_root.join("proto/google/crypto/tink/aes_gcm.proto"),
40
+
proto_root.join("google/api/annotations.proto"),
41
+
proto_root.join("google/api/client.proto"),
42
+
proto_root.join("google/api/field_behavior.proto"),
43
+
proto_root.join("google/api/http.proto"),
44
+
proto_root.join("google/api/resource.proto"),
45
+
proto_root.join("google/iam/v1/iam_policy.proto"),
46
+
proto_root.join("google/iam/v1/options.proto"),
47
+
proto_root.join("google/iam/v1/policy.proto"),
48
+
proto_root.join("google/longrunning/operations.proto"),
49
+
proto_root.join("google/rpc/status.proto"),
50
+
proto_root.join("google/type/expr.proto"),
51
+
proto_root.join("google/pubsub/v1/pubsub.proto"),
52
+
proto_root.join("google/cloud/kms/v1/service.proto"),
53
+
proto_root.join("google/cloud/kms/v1/resources.proto"),
54
+
proto_root.join("google/crypto/tink/aes_gcm.proto"),
55
55
],
56
56
&[proto_root],
57
57
)?;
58
58
59
59
tonic_build::configure().compile(
60
-
&[proto_root.join("proto/etcd_watcherd/v1/etcd_watcherd.proto")],
60
+
&[proto_root.join("etcd_watcherd/v1/etcd_watcherd.proto")],
61
61
&[proto_root],
62
62
)?;
63
63
+1
-1
osprey_coordinator/etcd_config_derive/Cargo.toml
+1
-1
osprey_coordinator/etcd_config_derive/Cargo.toml
+1
-1
osprey_coordinator/metrics_derive/Cargo.toml
+1
-1
osprey_coordinator/metrics_derive/Cargo.toml
+2
-2
osprey_coordinator/src/coordinator_metrics.rs
+2
-2
osprey_coordinator/src/coordinator_metrics.rs
···
1
1
use crate::metrics::define_metrics;
2
2
3
-
define_metrics!(SmiteCoordinatorMetrics, [
3
+
define_metrics!(OspreyCoordinatorMetrics, [
4
4
// How long an action has been processed by a worker for before it is acked/nacked
5
5
action_outstanding_duration => StaticHistogram(),
6
6
// How long an action has been held in the async queue for before it is sent to a worker
···
13
13
priority_queue_size_async => StaticGauge("priority_queue_size",["type" => "async"]),
14
14
15
15
// How many receivers are open for the priority queue
16
-
// can be used as a proxy for number of connections open from the smite worker
16
+
// can be used as a proxy for number of connections open from the osprey worker
17
17
priority_queue_receiver_count_sync => StaticGauge("priority_queue_receiver_count",["type" => "sync"]),
18
18
priority_queue_receiver_count_async => StaticGauge("priority_queue_receiver_count",["type" => "async"]),
19
19
+8
-8
osprey_coordinator/src/main.rs
+8
-8
osprey_coordinator/src/main.rs
···
17
17
mod pubsub;
18
18
mod shutdown_handler;
19
19
mod signals;
20
-
mod smite_bidirectional_stream;
20
+
mod osprey_bidirectional_stream;
21
21
mod snowflake_client;
22
22
mod sync_action_rpc;
23
23
mod tokio_utils;
···
25
25
mod tonic_mock;
26
26
use anyhow::Result;
27
27
use clap::Parser;
28
-
use proto::smite_coordinator_sync_action::smite_coordinator_sync_action_service_server::SmiteCoordinatorSyncActionServiceServer;
28
+
use proto::osprey_coordinator_sync_action::osprey_coordinator_sync_action_service_server::OspreyCoordinatorSyncActionServiceServer;
29
29
use std::sync::Arc;
30
30
use std::time::Duration;
31
31
32
32
use crate::snowflake_client::SnowflakeClient;
33
33
use crate::{
34
-
coordinator_metrics::SmiteCoordinatorMetrics, label_service_client::LabelServiceClient,
34
+
coordinator_metrics::OspreyCoordinatorMetrics, label_service_client::LabelServiceClient,
35
35
};
36
36
37
37
use crate::metrics::emit_worker::SpawnEmitWorker;
···
41
41
use pubsub::start_pubsub_subscriber;
42
42
use tokio::join;
43
43
44
-
use crate::proto::smite_coordinator_service_server::SmiteCoordinatorServiceServer;
45
-
use crate::smite_bidirectional_stream::SmiteCoordinatorServer;
44
+
use crate::proto::osprey_coordinator_service_server::OspreyCoordinatorServiceServer;
45
+
use crate::osprey_bidirectional_stream::OspreyCoordinatorServer;
46
46
47
47
#[derive(Debug, Parser)]
48
48
struct CliOptions {
···
78
78
let snowflake_client = Arc::new(SnowflakeClient::new(opts.snowflake_api_endpoint));
79
79
80
80
let (priority_queue_sender, priority_queue_receiver) = create_ackable_action_priority_queue();
81
-
let metrics = SmiteCoordinatorMetrics::new();
81
+
let metrics = OspreyCoordinatorMetrics::new();
82
82
tracing::info!("starting grpc metrics worker");
83
83
let _worker_guard = metrics
84
84
.clone()
85
85
.spawn_emit_worker(new_client("smite_coordinator").unwrap());
86
86
87
87
let smite_coordinator_grpc_bidi_stream_service =
88
-
SmiteCoordinatorServiceServer::new(SmiteCoordinatorServer::new(
88
+
OspreyCoordinatorServiceServer::new(OspreyCoordinatorServer::new(
89
89
priority_queue_sender.clone(),
90
90
priority_queue_receiver.clone(),
91
91
metrics.clone(),
···
95
95
let label_service_client = LabelServiceClient::new().await?;
96
96
97
97
let smite_coordinator_sync_action_service =
98
-
SmiteCoordinatorSyncActionServiceServer::new(sync_action_rpc::SyncActionServer::new(
98
+
OspreyCoordinatorSyncActionServiceServer::new(sync_action_rpc::SyncActionServer::new(
99
99
snowflake_client.clone(),
100
100
priority_queue_sender.clone(),
101
101
metrics.clone(),
+6
-6
osprey_coordinator/src/priority_queue.rs
+6
-6
osprey_coordinator/src/priority_queue.rs
···
5
5
time::{interval, Duration, Instant, MissedTickBehavior},
6
6
};
7
7
8
-
use crate::{coordinator_metrics::SmiteCoordinatorMetrics, proto};
8
+
use crate::{coordinator_metrics::OspreyCoordinatorMetrics, proto};
9
9
10
10
use crate::tokio_utils::AbortOnDrop;
11
11
use std::{cell::Cell, sync::Arc};
···
26
26
}
27
27
28
28
pub struct AckableAction {
29
-
pub action: proto::SmiteCoordinatorAction,
29
+
pub action: proto::OspreyCoordinatorAction,
30
30
acking_oneshot_sender: oneshot::Sender<AckOrNack>,
31
31
local_retry_count: Cell<u32>,
32
32
pub created_at: Instant,
···
34
34
35
35
impl AckableAction {
36
36
pub fn new(
37
-
action: proto::SmiteCoordinatorAction,
37
+
action: proto::OspreyCoordinatorAction,
38
38
) -> (
39
39
AckableAction,
40
40
oneshot::Receiver<crate::priority_queue::AckOrNack>,
···
49
49
(ackable_action, acking_oneshot_receiver)
50
50
}
51
51
52
-
pub fn into_action(self) -> (proto::SmiteCoordinatorAction, ActionAcker) {
52
+
pub fn into_action(self) -> (proto::OspreyCoordinatorAction, ActionAcker) {
53
53
(
54
54
self.action,
55
55
ActionAcker {
···
167
167
}
168
168
pub async fn recv(
169
169
&self,
170
-
metrics: Arc<SmiteCoordinatorMetrics>,
170
+
metrics: Arc<OspreyCoordinatorMetrics>,
171
171
) -> Result<AckableAction, async_channel::RecvError> {
172
172
loop {
173
173
let result = tokio::select! {
···
223
223
224
224
pub fn spawn_priority_queue_metrics_worker(
225
225
queue_sender: PriorityQueueSender,
226
-
metrics: Arc<SmiteCoordinatorMetrics>,
226
+
metrics: Arc<OspreyCoordinatorMetrics>,
227
227
) -> AbortOnDrop<()> {
228
228
let mut interval = interval(Duration::from_millis(100));
229
229
interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+32
-48
osprey_coordinator/src/proto/mod.rs
+32
-48
osprey_coordinator/src/proto/mod.rs
···
1
1
#![allow(unused_imports)]
2
2
3
-
pub mod discord_smite_rpc {
4
-
pub mod actions {
5
-
pub mod v1 {
6
-
include!(concat!(env!("OUT_DIR"), "/discord_smite_rpc.actions.v1.rs"));
7
-
}
8
-
}
9
-
10
-
pub mod common {
11
-
pub mod v1 {
12
-
include!(concat!(env!("OUT_DIR"), "/discord_smite_rpc.common.v1.rs"));
13
-
}
14
-
}
15
-
16
-
pub mod labels {
17
-
pub mod v1 {
18
-
include!(concat!(env!("OUT_DIR"), "/discord_smite_rpc.labels.v1.rs"));
3
+
pub mod osprey {
4
+
pub mod rpc {
5
+
pub mod actions {
6
+
pub mod v1 {
7
+
include!(concat!(env!("OUT_DIR"), "/osprey.rpc.actions.v1.rs"));
8
+
}
19
9
}
20
-
}
21
-
22
-
pub mod smite_coordinator {
23
-
pub mod bidirectional_stream {
10
+
pub mod common {
24
11
pub mod v1 {
25
-
include!(concat!(
26
-
env!("OUT_DIR"),
27
-
"/discord_smite_rpc.smite_coordinator.bidirectional_stream.v1.rs"
28
-
));
12
+
include!(concat!(env!("OUT_DIR"), "/osprey.rpc.common.v1.rs"));
29
13
}
30
14
}
31
15
32
-
pub mod sync_action {
16
+
pub mod labels {
33
17
pub mod v1 {
34
-
include!(concat!(
35
-
env!("OUT_DIR"),
36
-
"/discord_smite_rpc.smite_coordinator.sync_action.v1.rs"
37
-
));
18
+
include!(concat!(env!("OUT_DIR"), "/osprey.rpc.labels.v1.rs"));
38
19
}
39
20
}
40
-
}
41
-
}
42
21
43
-
pub mod discord_protos {
44
-
pub mod discord_authentication {
45
-
pub mod v1 {
46
-
include!(concat!(
47
-
env!("OUT_DIR"),
48
-
"/discord_protos.discord_authentication.v1.rs"
49
-
));
50
-
}
51
-
}
22
+
pub mod osprey_coordinator {
23
+
pub mod bidirectional_stream {
24
+
pub mod v1 {
25
+
include!(concat!(
26
+
env!("OUT_DIR"),
27
+
"/osprey.rpc.osprey_coordinator.bidirectional_stream.v1.rs"
28
+
));
29
+
}
30
+
}
52
31
53
-
pub mod users {
54
-
pub mod v1 {
55
-
include!(concat!(env!("OUT_DIR"), "/discord_protos.users.v1.rs"));
32
+
pub mod sync_action {
33
+
pub mod v1 {
34
+
include!(concat!(
35
+
env!("OUT_DIR"),
36
+
"/osprey.rpc.osprey_coordinator.sync_action.v1.rs"
37
+
));
38
+
}
39
+
}
56
40
}
57
41
}
58
42
}
59
43
60
-
pub use discord_smite_rpc::actions::v1::*;
61
-
pub use discord_smite_rpc::common::v1::*;
62
-
pub use discord_smite_rpc::labels::v1::*;
63
-
pub use discord_smite_rpc::smite_coordinator::bidirectional_stream::v1::*;
64
-
pub use discord_smite_rpc::smite_coordinator::sync_action::v1 as smite_coordinator_sync_action;
65
44
45
+
pub use osprey::rpc::actions::v1::*;
46
+
pub use osprey::rpc::common::v1::*;
47
+
pub use osprey::rpc::labels::v1::*;
48
+
pub use osprey::rpc::osprey_coordinator::bidirectional_stream::v1::*;
49
+
pub use osprey::rpc::osprey_coordinator::sync_action::v1 as osprey_coordinator_sync_action;
66
50
pub const PB_DESCRIPTOR_BYTES: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/descriptor.bin"));
+12
-12
osprey_coordinator/src/pubsub.rs
+12
-12
osprey_coordinator/src/pubsub.rs
···
14
14
use crate::metrics::histograms::StaticHistogram;
15
15
use crate::metrics::MetricsClientBuilder;
16
16
use crate::{
17
-
coordinator_metrics::SmiteCoordinatorMetrics,
17
+
coordinator_metrics::OspreyCoordinatorMetrics,
18
18
priority_queue::{AckOrNack, AckableAction, PriorityQueueSender},
19
-
proto::{self, smite_coordinator_action::SecretData},
19
+
proto::{self, osprey_coordinator_action::SecretData},
20
20
pub_sub_streaming_pull::DetachedMessage,
21
21
pub_sub_streaming_pull::{FlowControl, SpawnTaskPerMessageHandler, StreamingPullManager},
22
22
};
···
34
34
use crate::signals::exit_signal;
35
35
use crate::snowflake_client::SnowflakeClient;
36
36
use convert_case::{Case, Casing};
37
-
use proto::smite_coordinator_action::ActionData;
37
+
use proto::osprey_coordinator_action::ActionData;
38
38
39
39
async fn decode_proto_message(
40
40
message_data: &[u8],
41
41
ack_id: u64,
42
42
message_timestamp: Timestamp,
43
43
snowflake_client: &SnowflakeClient,
44
-
metrics: &SmiteCoordinatorMetrics,
45
-
) -> Result<proto::SmiteCoordinatorAction> {
44
+
metrics: &OspreyCoordinatorMetrics,
45
+
) -> Result<proto::OspreyCoordinatorAction> {
46
46
let smite_proto_action = SmiteProtoAction::decode(message_data).unwrap();
47
47
let action_id = if smite_proto_action.id == 0 {
48
48
metrics.action_id_snowflake_generation_proto.incr();
···
55
55
.unwrap()
56
56
.to_string()
57
57
.to_case(Case::Snake);
58
-
Ok(proto::SmiteCoordinatorAction {
58
+
Ok(proto::OspreyCoordinatorAction {
59
59
ack_id,
60
60
action_id,
61
61
action_name,
···
70
70
ack_id: u64,
71
71
message_timestamp: Timestamp,
72
72
snowflake_client: &SnowflakeClient,
73
-
metrics: &SmiteCoordinatorMetrics,
74
-
) -> Result<proto::SmiteCoordinatorAction> {
73
+
metrics: &OspreyCoordinatorMetrics,
74
+
) -> Result<proto::OspreyCoordinatorAction> {
75
75
// This whole function can probably be optimized way better, but in the interest of time I am leaving
76
76
// it in a working state for now.
77
77
#[derive(Deserialize, Debug)]
···
106
106
}
107
107
};
108
108
109
-
Ok(proto::SmiteCoordinatorAction {
109
+
Ok(proto::OspreyCoordinatorAction {
110
110
ack_id,
111
111
action_id,
112
112
action_name: pubsub_action.name,
···
133
133
ack_id: u64,
134
134
message_timestamp: Timestamp,
135
135
snowflake_client: &SnowflakeClient,
136
-
metrics: &SmiteCoordinatorMetrics,
137
-
) -> Result<proto::SmiteCoordinatorAction> {
136
+
metrics: &OspreyCoordinatorMetrics,
137
+
) -> Result<proto::OspreyCoordinatorAction> {
138
138
let decrypted_message_vector = match message_attributes.get("encrypted") {
139
139
Some(is_encrypted) if is_encrypted == "true" => {
140
140
Some(decrypt_pubsub_message(kms_envelope, message_data).await?)
···
202
202
pub async fn start_pubsub_subscriber(
203
203
snowflake_client: Arc<SnowflakeClient>,
204
204
priority_queue_sender: PriorityQueueSender,
205
-
metrics: Arc<SmiteCoordinatorMetrics>,
205
+
metrics: Arc<OspreyCoordinatorMetrics>,
206
206
) -> Result<()> {
207
207
let subscriber_client = create_pubsub_subscription_client().await;
208
208
let subscription_name = if std::env::var("PUBSUB_EMULATOR_HOST").is_ok() {
+24
-24
osprey_coordinator/src/smite_bidirectional_stream.rs
osprey_coordinator/src/osprey_bidirectional_stream.rs
+24
-24
osprey_coordinator/src/smite_bidirectional_stream.rs
osprey_coordinator/src/osprey_bidirectional_stream.rs
···
1
-
use crate::coordinator_metrics::SmiteCoordinatorMetrics;
1
+
use crate::coordinator_metrics::OspreyCoordinatorMetrics;
2
2
use crate::priority_queue::ActionAcker;
3
3
use crate::priority_queue::{PriorityQueueReceiver, PriorityQueueSender};
4
4
use crate::proto;
···
49
49
OutstandingAction(OutstandingActionState),
50
50
}
51
51
52
-
pub struct SmiteCoordinatorServer {
52
+
pub struct OspreyCoordinatorServer {
53
53
priority_queue_receiver: PriorityQueueReceiver,
54
54
#[allow(unused)]
55
55
priority_queue_sender: PriorityQueueSender, // TODO: use this for retrying sync actions
56
-
metrics: Arc<SmiteCoordinatorMetrics>,
56
+
metrics: Arc<OspreyCoordinatorMetrics>,
57
57
}
58
58
59
-
impl SmiteCoordinatorServer {
59
+
impl OspreyCoordinatorServer {
60
60
pub fn new(
61
61
priority_queue_sender: PriorityQueueSender,
62
62
priority_queue_receiver: PriorityQueueReceiver,
63
-
metrics: Arc<SmiteCoordinatorMetrics>,
64
-
) -> SmiteCoordinatorServer {
65
-
SmiteCoordinatorServer {
63
+
metrics: Arc<OspreyCoordinatorMetrics>,
64
+
) -> OspreyCoordinatorServer {
65
+
OspreyCoordinatorServer {
66
66
priority_queue_sender,
67
67
priority_queue_receiver,
68
68
metrics,
···
73
73
fn handle_action_request(
74
74
action_request: ActionRequest,
75
75
current_client_state: ClientState,
76
-
metrics: Arc<SmiteCoordinatorMetrics>,
76
+
metrics: Arc<OspreyCoordinatorMetrics>,
77
77
) -> Result<proto::ClientDetails> {
78
78
match (action_request, current_client_state) {
79
79
(ActionRequest::Initial(client_details), ClientState::NoOutstandingAction) => {
···
109
109
110
110
async fn handle_request(
111
111
client_state: ClientState,
112
-
sender: &Sender<Result<proto::SmiteCoordinatorAction, tonic::Status>>,
112
+
sender: &Sender<Result<proto::OspreyCoordinatorAction, tonic::Status>>,
113
113
request: proto::Request,
114
114
action_receiver: &PriorityQueueReceiver,
115
-
metrics: Arc<SmiteCoordinatorMetrics>,
115
+
metrics: Arc<OspreyCoordinatorMetrics>,
116
116
receive_timeout: Duration,
117
117
) -> Result<UpdateClientStateOrDisconnect> {
118
118
match request
···
168
168
}
169
169
170
170
#[tonic::async_trait]
171
-
impl proto::smite_coordinator_service_server::SmiteCoordinatorService for SmiteCoordinatorServer {
172
-
type SmiteBidirectionalStreamStream =
173
-
ReceiverStream<Result<proto::SmiteCoordinatorAction, tonic::Status>>;
171
+
impl proto::osprey_coordinator_service_server::OspreyCoordinatorService for OspreyCoordinatorServer {
172
+
type OspreyBidirectionalStreamStream =
173
+
ReceiverStream<Result<proto::OspreyCoordinatorAction, tonic::Status>>;
174
174
175
-
async fn smite_bidirectional_stream(
175
+
async fn osprey_bidirectional_stream(
176
176
&self,
177
177
request: tonic::Request<tonic::Streaming<proto::Request>>,
178
-
) -> Result<tonic::Response<Self::SmiteBidirectionalStreamStream>, tonic::Status> {
178
+
) -> Result<tonic::Response<Self::OspreyBidirectionalStreamStream>, tonic::Status> {
179
179
tracing::debug!(
180
180
{ connection =? request.metadata() },
181
181
"New Connection Received"
···
286
286
#[cfg(test)]
287
287
mod tests {
288
288
289
-
use crate::coordinator_metrics::SmiteCoordinatorMetrics;
289
+
use crate::coordinator_metrics::OspreyCoordinatorMetrics;
290
290
use crate::metrics::emit_worker::SpawnEmitWorker;
291
291
use crate::metrics::new_client;
292
-
use crate::proto::smite_coordinator_action::ActionData;
293
-
use crate::proto::smite_coordinator_action::SecretData;
294
-
use proto::smite_coordinator_service_server::SmiteCoordinatorService;
292
+
use crate::proto::osprey_coordinator_action::ActionData;
293
+
use crate::proto::osprey_coordinator_action::SecretData;
294
+
use proto::osprey_coordinator_service_server::OspreyCoordinatorService;
295
295
296
296
use crate::priority_queue::create_ackable_action_priority_queue;
297
297
use crate::priority_queue::AckableAction;
···
306
306
tracing_subscriber::fmt::init();
307
307
let (priority_queue_sender, priority_queue_receiver) =
308
308
create_ackable_action_priority_queue();
309
-
let metrics = SmiteCoordinatorMetrics::new();
309
+
let metrics = OspreyCoordinatorMetrics::new();
310
310
let _worker_guard = metrics
311
311
.clone()
312
312
.spawn_emit_worker(new_client("smite_coordinator").unwrap());
313
313
314
-
let ackable_action = proto::SmiteCoordinatorAction {
314
+
let ackable_action = proto::OspreyCoordinatorAction {
315
315
ack_id: 1,
316
316
action_id: 1,
317
317
action_name: "test_action".into(),
···
329
329
.await
330
330
.unwrap();
331
331
332
-
let ackable_action_2 = proto::SmiteCoordinatorAction {
332
+
let ackable_action_2 = proto::OspreyCoordinatorAction {
333
333
ack_id: 2,
334
334
action_id: 2,
335
335
action_name: "test_action".into(),
···
347
347
.await
348
348
.unwrap();
349
349
350
-
let server = SmiteCoordinatorServer::new(
350
+
let server = OspreyCoordinatorServer::new(
351
351
priority_queue_sender.clone(),
352
352
priority_queue_receiver,
353
353
metrics.clone(),
···
385
385
]);
386
386
387
387
let res = server
388
-
.smite_bidirectional_stream(req)
388
+
.osprey_bidirectional_stream(req)
389
389
.await
390
390
.expect("error in stream");
391
391
+16
-16
osprey_coordinator/src/sync_action_rpc.rs
+16
-16
osprey_coordinator/src/sync_action_rpc.rs
···
2
2
use crate::metrics::histograms::StaticHistogram;
3
3
use crate::snowflake_client::SnowflakeClient;
4
4
use crate::{
5
-
coordinator_metrics::SmiteCoordinatorMetrics,
5
+
coordinator_metrics::OspreyCoordinatorMetrics,
6
6
label_service_client::LabelServiceClient,
7
7
priority_queue::AckableAction,
8
8
priority_queue::{AckOrNack, PriorityQueueSender},
9
-
proto::{self, smite_coordinator_sync_action},
9
+
proto::{self, osprey_coordinator_sync_action},
10
10
};
11
11
use anyhow::{anyhow, Context, Result};
12
12
use std::sync::Arc;
13
13
use tokio::time::Instant;
14
14
15
15
use rand::Rng;
16
-
use smite_coordinator_sync_action::smite_coordinator_sync_action_service_server::SmiteCoordinatorSyncActionService;
17
-
use smite_coordinator_sync_action::ProcessActionRequest;
16
+
use osprey_coordinator_sync_action::osprey_coordinator_sync_action_service_server::OspreyCoordinatorSyncActionService;
17
+
use osprey_coordinator_sync_action::ProcessActionRequest;
18
18
19
19
pub(crate) struct SyncActionServer {
20
20
snowflake_client: Arc<SnowflakeClient>,
21
21
priority_queue_sender: PriorityQueueSender,
22
-
metrics: Arc<SmiteCoordinatorMetrics>,
22
+
metrics: Arc<OspreyCoordinatorMetrics>,
23
23
label_service_client: LabelServiceClient,
24
24
}
25
25
···
27
27
pub fn new(
28
28
snowflake_client: Arc<SnowflakeClient>,
29
29
priority_queue_sender: PriorityQueueSender,
30
-
metrics: Arc<SmiteCoordinatorMetrics>,
30
+
metrics: Arc<OspreyCoordinatorMetrics>,
31
31
label_service_client: LabelServiceClient,
32
32
) -> SyncActionServer {
33
33
SyncActionServer {
···
41
41
42
42
async fn create_smite_coordinator_action(
43
43
ack_id: u64,
44
-
action_request: &smite_coordinator_sync_action::ProcessActionRequest,
44
+
action_request: &osprey_coordinator_sync_action::ProcessActionRequest,
45
45
snowflake_client: &SnowflakeClient,
46
-
) -> Result<(proto::SmiteCoordinatorAction, Vec<crate::proto::EntityKey>)> {
46
+
) -> Result<(proto::OspreyCoordinatorAction, Vec<crate::proto::EntityKey>)> {
47
47
// generate snowflake if one is not provided, to match the behaviour in pubsub.rs
48
48
let action_id = match action_request.action_id {
49
49
Some(id) => match id {
···
56
56
if action_request.action_name.is_empty() {
57
57
return Err(anyhow!("`action_name` must not be empty"));
58
58
}
59
-
let smite_coordinator_action = proto::SmiteCoordinatorAction {
59
+
let smite_coordinator_action = proto::OspreyCoordinatorAction {
60
60
ack_id,
61
61
action_id,
62
62
action_name: action_request.action_name.clone(),
63
-
action_data: Some(proto::smite_coordinator_action::ActionData::JsonActionData(
63
+
action_data: Some(proto::osprey_coordinator_action::ActionData::JsonActionData(
64
64
action_request.action_data_json.clone().into(),
65
65
)),
66
66
secret_data: None,
···
84
84
&self,
85
85
ack_id: u64,
86
86
action_request: &ProcessActionRequest,
87
-
) -> Result<tonic::Response<smite_coordinator_sync_action::ProcessActionResponse>, tonic::Status>
87
+
) -> Result<tonic::Response<osprey_coordinator_sync_action::ProcessActionResponse>, tonic::Status>
88
88
{
89
89
let unvalidated_action_id = action_request.action_id;
90
90
91
-
let (smite_coordinator_action, requested_entities) = match create_smite_coordinator_action(
91
+
let (smite_coordinator_action, requested_entities) = match create_smite_coordinator_action(
92
92
ack_id,
93
93
action_request,
94
94
self.snowflake_client.as_ref(),
···
146
146
};
147
147
148
148
let response =
149
-
smite_coordinator_sync_action::ProcessActionResponse { entities, verdicts };
149
+
osprey_coordinator_sync_action::ProcessActionResponse { entities, verdicts };
150
150
151
151
self.metrics.sync_classification_result_ack.incr();
152
152
self.metrics
···
178
178
}
179
179
180
180
#[tonic::async_trait]
181
-
impl SmiteCoordinatorSyncActionService for SyncActionServer {
181
+
impl OspreyCoordinatorSyncActionService for SyncActionServer {
182
182
async fn process_action(
183
183
&self,
184
-
request: tonic::Request<smite_coordinator_sync_action::ProcessActionRequest>,
185
-
) -> Result<tonic::Response<smite_coordinator_sync_action::ProcessActionResponse>, tonic::Status>
184
+
request: tonic::Request<osprey_coordinator_sync_action::ProcessActionRequest>,
185
+
) -> Result<tonic::Response<osprey_coordinator_sync_action::ProcessActionResponse>, tonic::Status>
186
186
{
187
187
self.metrics.sync_classification_action_received.incr();
188
188
let action_request = request.into_inner();
+5
-7
proto/osprey/rpc/actions/v1/action.proto
+5
-7
proto/osprey/rpc/actions/v1/action.proto
···
1
1
syntax = "proto3";
2
2
3
-
package discord_smite_rpc.actions.v1;
3
+
package osprey.rpc.actions.v1;
4
4
5
-
import "discord_protos/discord_authentication/v1/session.proto";
6
-
import "discord_smite_rpc/actions/v1/action_types.proto";
5
+
6
+
import "osprey/rpc/actions/v1/action_types.proto";
7
7
import "google/protobuf/timestamp.proto";
8
8
import "google/protobuf/wrappers.proto";
9
9
···
140
140
}
141
141
142
142
message UserSession {
143
+
reserved 2, 6, 7;
144
+
143
145
bytes id_hash = 1;
144
-
discord_protos.discord_authentication.v1.UserSessionData.Version session_version = 2;
145
146
google.protobuf.Timestamp creation_time = 3;
146
147
google.protobuf.Timestamp expiration_time = 4;
147
148
google.protobuf.Timestamp approx_last_used_time = 5;
148
-
discord_protos.discord_authentication.v1.UserSessionData.ClientInfo client_info = 6;
149
-
// Previous client info of the session if a change was detected
150
-
discord_protos.discord_authentication.v1.UserSessionData.ClientInfo prev_client_info = 7;
151
149
google.protobuf.Timestamp prev_approx_last_used_time = 8;
152
150
}
153
151
+8
-8
proto/osprey/rpc/actions/v1/action_types.proto
+8
-8
proto/osprey/rpc/actions/v1/action_types.proto
···
1
1
syntax = "proto3";
2
2
3
-
package discord_smite_rpc.actions.v1;
3
+
package osprey.rpc.actions.v1;
4
4
5
-
import "discord_smite_rpc/actions/v1/application.proto";
6
-
import "discord_smite_rpc/actions/v1/captcha.proto";
7
-
import "discord_smite_rpc/actions/v1/channel.proto";
8
-
import "discord_smite_rpc/actions/v1/guild.proto";
9
-
import "discord_smite_rpc/actions/v1/invite.proto";
10
-
import "discord_smite_rpc/actions/v1/metadata.proto";
11
-
import "discord_smite_rpc/actions/v1/user.proto";
5
+
import "osprey/rpc/actions/v1/application.proto";
6
+
import "osprey/rpc/actions/v1/captcha.proto";
7
+
import "osprey/rpc/actions/v1/channel.proto";
8
+
import "osprey/rpc/actions/v1/guild.proto";
9
+
import "osprey/rpc/actions/v1/invite.proto";
10
+
import "osprey/rpc/actions/v1/metadata.proto";
11
+
import "osprey/rpc/actions/v1/user.proto";
12
12
13
13
message GuildCreated {
14
14
User user = 1;
+2
-2
proto/osprey/rpc/actions/v1/application.proto
+2
-2
proto/osprey/rpc/actions/v1/application.proto
···
1
1
syntax = "proto3";
2
2
3
-
package discord_smite_rpc.actions.v1;
3
+
package osprey.rpc.actions.v1;
4
4
5
-
import "discord_smite_rpc/actions/v1/user.proto";
5
+
import "osprey/rpc/actions/v1/user.proto";
6
6
7
7
import "google/protobuf/timestamp.proto";
8
8
import "google/protobuf/wrappers.proto";
+1
-1
proto/osprey/rpc/actions/v1/captcha.proto
+1
-1
proto/osprey/rpc/actions/v1/captcha.proto
+1
-1
proto/osprey/rpc/actions/v1/channel.proto
+1
-1
proto/osprey/rpc/actions/v1/channel.proto
+1
-1
proto/osprey/rpc/actions/v1/guild.proto
+1
-1
proto/osprey/rpc/actions/v1/guild.proto
+1
-1
proto/osprey/rpc/actions/v1/invite.proto
+1
-1
proto/osprey/rpc/actions/v1/invite.proto
+1
-1
proto/osprey/rpc/actions/v1/metadata.proto
+1
-1
proto/osprey/rpc/actions/v1/metadata.proto
+2
-4
proto/osprey/rpc/actions/v1/user.proto
+2
-4
proto/osprey/rpc/actions/v1/user.proto
···
1
1
syntax = "proto3";
2
2
3
-
package discord_smite_rpc.actions.v1;
3
+
package osprey.rpc.actions.v1;
4
4
5
-
import "discord_protos/users/v1/user.proto";
6
5
import "google/protobuf/timestamp.proto";
7
6
import "google/protobuf/wrappers.proto";
8
7
9
8
message User {
10
-
reserved 10;
9
+
reserved 10, 23;
11
10
12
11
fixed64 id = 1;
13
12
bool mfa_enabled = 2;
···
31
30
google.protobuf.StringValue global_name = 21;
32
31
// Used by guild automation GUILD_JOINED event
33
32
google.protobuf.StringValue hub_email = 22;
34
-
discord_protos.users.v1.UserAvatarDecoration avatar_decoration_data = 23;
35
33
}
+36
proto/osprey/rpc/labels/v1/service.proto
+36
proto/osprey/rpc/labels/v1/service.proto
···
1
+
syntax = "proto3";
2
+
3
+
package osprey.rpc.labels.v1;
4
+
5
+
// Mock proto file for deprecated labels functionality
6
+
// These are marked as DEPRECATED in sync_action service
7
+
// and will be removed once labels are replaced with verdicts
8
+
9
+
message EntityKey {
10
+
string entity_type = 1;
11
+
string entity_id = 2;
12
+
}
13
+
14
+
message Label {
15
+
string name = 1;
16
+
int64 version = 2;
17
+
int64 expires_at = 3;
18
+
}
19
+
20
+
message Entity {
21
+
EntityKey key = 1;
22
+
repeated Label labels = 2;
23
+
}
24
+
25
+
message GetEntityRequest {
26
+
string routing_key = 1;
27
+
EntityKey key = 2;
28
+
}
29
+
30
+
message GetEntityResponse {
31
+
Entity entity = 1;
32
+
}
33
+
34
+
service LabelService {
35
+
rpc GetEntity(GetEntityRequest) returns (GetEntityResponse);
36
+
}