Skip to content

Adding metadata to shardcake's internal calls. #155

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

Merged
ghostdogpr merged 10 commits into from
Apr 12, 2025
Merged

Adding metadata to shardcake's internal calls. #155

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0754af8
Example of adding metadata to shardcake's internal calls.
douglasthomsen Mar 31, 2025
5666b73
Updating var name.
douglasthomsen Mar 31, 2025
f01a7bc
Renaming GrpcConfig vars
douglasthomsen Apr 1, 2025
f3f0d1b
Moving HandlerAspect to an argument in Server.run.
douglasthomsen Apr 1, 2025
f3cd4a0
Updating comments.
douglasthomsen Apr 1, 2025
85d0c61
Removing unused changes.
douglasthomsen Apr 1, 2025
aec52d8
Adding serverInterceptors implementation.
douglasthomsen Apr 2, 2025
4f79abf
Adding tests
douglasthomsen Apr 11, 2025
4f1627b
Merge branch 'series/2.x' into add-request-metadata
douglasthomsen Apr 11, 2025
15e94e1
Adding tests
douglasthomsen Apr 11, 2025
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
7 changes: 5 additions & 2 deletions entities/src/main/scala/com/devsisters/shardcake/Config.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package com.devsisters.shardcake

import sttp.client3.UriContext
import sttp.model.Header
import sttp.model.Uri
import zio._

Expand Down Expand Up @@ -29,7 +30,8 @@ case class Config(
sendTimeout: Duration,
refreshAssignmentsRetryInterval: Duration,
unhealthyPodReportInterval: Duration,
simulateRemotePods: Boolean
simulateRemotePods: Boolean,
managerClientHeaderInterceptor: Seq[Header] => Seq[Header]
)

object Config {
Expand All @@ -44,6 +46,7 @@ object Config {
sendTimeout = 10 seconds,
refreshAssignmentsRetryInterval = 5 seconds,
unhealthyPodReportInterval = 5 seconds,
simulateRemotePods = false
simulateRemotePods = false,
managerClientHeaderInterceptor = headers => headers
)
}
6 changes: 4 additions & 2 deletions entities/src/main/scala/com/devsisters/shardcake/ShardManagerClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,10 @@ object ShardManagerClient {
}

class ShardManagerClientLive(sttp: SttpBackend[Task, Any], config: Config) extends ShardManagerClient {
private def send[Origin: IsOperation, A](query: SelectionBuilder[Origin, A]): Task[A] =
sttp.send(query.toRequest(config.shardManagerUri)).map(_.body).absolve
private def send[Origin: IsOperation, A](query: SelectionBuilder[Origin, A]): Task[A] = {
val request = query.toRequest(config.shardManagerUri)
sttp.send(request.copy(headers = config.managerClientHeaderInterceptor(request.headers))).map(_.body).absolve
}

def register(podAddress: PodAddress): Task[Unit] =
send(
Expand Down
7 changes: 5 additions & 2 deletions manager/src/main/scala/com/devsisters/shardcake/ManagerConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package com.devsisters.shardcake

import zio._
import zio.http.HandlerAspect

/**
* Shard Manager configuration
Expand All @@ -23,7 +24,8 @@ case class ManagerConfig(
persistRetryInterval: Duration,
persistRetryCount: Int,
rebalanceRate: Double,
podHealthCheckInterval: Duration
podHealthCheckInterval: Duration,
httpHandler: HandlerAspect[Any, Unit]
)

object ManagerConfig {
Expand All @@ -37,6 +39,7 @@ object ManagerConfig {
persistRetryInterval = 3 seconds,
persistRetryCount = 100,
rebalanceRate = 2 / 100d,
podHealthCheckInterval = 1 minute
podHealthCheckInterval = 1 minute,
httpHandler = HandlerAspect.identity
)
}
4 changes: 2 additions & 2 deletions manager/src/main/scala/com/devsisters/shardcake/Server.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@ object Server {
handlers = QuickAdapter(interpreter).handlers
routes = Routes(
Method.ANY / "health" -> Handler.ok,
Method.ANY / "api" / "graphql" -> handlers.api,
Method.ANY / "ws" / "graphql" -> handlers.webSocket
Method.ANY / "api" / "graphql" -> handlers.api @@ config.httpHandler,
Method.ANY / "ws" / "graphql" -> handlers.webSocket @@ config.httpHandler
) @@ Middleware.cors
_ <- ZIO.logInfo(s"Shard Manager server started on port ${config.apiPort}.")
nothing <- ZServer
Expand Down
8 changes: 6 additions & 2 deletions protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
package com.devsisters.shardcake

import zio._
import io.grpc.StatusException
import scalapb.zio_grpc.GTransform
import scalapb.zio_grpc.ZClientInterceptor

import java.util.concurrent.Executor
import scalapb.zio_grpc.RequestContext

/**
* The configuration for the gRPC client.
Expand All @@ -17,10 +20,11 @@ case class GrpcConfig(
maxInboundMessageSize: Int,
executor: Option[Executor],
shutdownTimeout: Duration,
interceptors: Seq[ZClientInterceptor]
interceptors: Seq[ZClientInterceptor],
serviceTransform: GTransform[RequestContext, StatusException, RequestContext, StatusException]
Copy link
Collaborator

@ghostdogpr ghostdogpr Apr 1, 2025

Choose a reason for hiding this comment

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

Looks ok, how about renaming interceptors to clientInterceptors and serviceTransform to serverInterceptors?

Even though identity (and similarly the empty aspect) does nothing, a list might be a little better? So that we don't call anything additional when we don't need it.

Copy link
Contributor Author

@douglasthomsen douglasthomsen Apr 1, 2025

Choose a reason for hiding this comment

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

I have renamed the var names but I have not updated how the GrpcShardingService adds the serviceTransform to the service. I will work on this tomorrow.

Copy link
Contributor Author

@douglasthomsen douglasthomsen Apr 2, 2025

Choose a reason for hiding this comment

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

I have updated the code to add the serverInterceptors. I will work on adding some tests next.

Copy link
Contributor Author

@douglasthomsen douglasthomsen Apr 12, 2025

Choose a reason for hiding this comment

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

@ghostdogpr I have added some tests for adding metadata to shardcake's internal calls. I put them in the example subproject because it had access to all of the code that I needed but I am not sure if that is the right place for them.

Can you review these changes again? Also, is there any documentation that I need to update for this change?

)

object GrpcConfig {
val default: GrpcConfig =
GrpcConfig(maxInboundMessageSize = 32 * 1024 * 1024, None, 3.seconds, Seq.empty)
GrpcConfig(maxInboundMessageSize = 32 * 1024 * 1024, None, 3.seconds, Seq.empty, GTransform.identity)
}
5 changes: 4 additions & 1 deletion protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcShardingService.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,10 @@ object GrpcShardingService {
case None =>
ServerBuilder.forPort(config.shardingPort)
}
services <- ServiceList.add(new GrpcShardingService(sharding, config.sendTimeout) {}).bindAll
services <-
ServiceList
.add(new GrpcShardingService(sharding, config.sendTimeout) {}.transform(grpcConfig.serviceTransform))
.bindAll
server: Server = services
.foldLeft(builder) { case (builder0, service) => builder0.addService(service) }
.addService(ProtoReflectionService.newInstance())
Expand Down