[OPIK-68] Add rate limit

apps/opik-backend/config.yml

@@ -65,3 +65,9 @@ server:
   enableVirtualThreads: ${ENABLE_VIRTUAL_THREADS:-false}
   gzip:
     enabled: true
+
+rateLimit:
+    enabled: ${RATE_LIMIT_ENABLED:-false}
+    generalEvents:
+      limit: ${RATE_LIMIT_GENERAL_EVENTS_LIMIT:-5000}
+      durationInSeconds: ${RATE_LIMIT_GENERAL_EVENTS_DURATION_IN_SEC:-1}

apps/opik-backend/src/main/java/com/comet/opik/OpikApplication.java

@@ -6,6 +6,7 @@
 import com.comet.opik.infrastructure.db.DatabaseAnalyticsModule;
 import com.comet.opik.infrastructure.db.IdGeneratorModule;
 import com.comet.opik.infrastructure.db.NameGeneratorModule;
+import com.comet.opik.infrastructure.ratelimit.RateLimitModule;
 import com.comet.opik.infrastructure.redis.RedisModule;
 import com.comet.opik.utils.JsonBigDecimalDeserializer;
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -60,7 +61,7 @@ public void initialize(Bootstrap<OpikConfiguration> bootstrap) {
                 .bundles(JdbiBundle.<OpikConfiguration>forDatabase((conf, env) -> conf.getDatabase())
                         .withPlugins(new SqlObjectPlugin(), new Jackson2Plugin()))
                 .modules(new DatabaseAnalyticsModule(), new IdGeneratorModule(), new AuthModule(), new RedisModule(),
-                        new NameGeneratorModule())
+                        new RateLimitModule(), new NameGeneratorModule())
                 .enableAutoConfig()
                 .build());
     }

apps/opik-backend/src/main/java/com/comet/opik/api/DatasetItemBatch.java

@@ -1,6 +1,7 @@
 package com.comet.opik.api;
 
 import com.comet.opik.api.validate.DatasetItemBatchValidation;
+import com.comet.opik.infrastructure.ratelimit.RateEventContainer;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonView;
 import com.fasterxml.jackson.databind.PropertyNamingStrategies;
@@ -26,6 +27,12 @@ public record DatasetItemBatch(
                 DatasetItem.View.Write.class}) @Pattern(regexp = NULL_OR_NOT_BLANK, message = "must not be blank") @Schema(description = "If null, dataset_id must be provided") String datasetName,
         @JsonView({
                 DatasetItem.View.Write.class}) @Schema(description = "If null, dataset_name must be provided") UUID datasetId,
-        @JsonView({DatasetItem.View.Write.class}) @NotNull @Size(min = 1, max = 1000) @Valid List<DatasetItem> items){
+        @JsonView({DatasetItem.View.Write.class}) @NotNull @Size(min = 1, max = 1000) @Valid List<DatasetItem> items)
+        implements
+            RateEventContainer{
 
+    @Override
+    public long eventCount() {
+        return items.size();
+    }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/ExperimentItemsBatch.java

@@ -1,5 +1,6 @@
 package com.comet.opik.api;
 
+import com.comet.opik.infrastructure.ratelimit.RateEventContainer;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonView;
 import com.fasterxml.jackson.databind.PropertyNamingStrategies;
@@ -16,5 +17,12 @@
 @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
 public record ExperimentItemsBatch(
         @JsonView( {
-                ExperimentItem.View.Write.class}) @NotNull @Size(min = 1, max = 1000) @Valid Set<ExperimentItem> experimentItems){
+                ExperimentItem.View.Write.class}) @NotNull @Size(min = 1, max = 1000) @Valid Set<ExperimentItem> experimentItems)
+        implements
+            RateEventContainer{
+
+    @Override
+    public long eventCount() {
+        return experimentItems.size();
+    }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/FeedbackScoreBatch.java

@@ -1,5 +1,6 @@
 package com.comet.opik.api;
 
+import com.comet.opik.infrastructure.ratelimit.RateEventContainer;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.databind.PropertyNamingStrategies;
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
@@ -13,6 +14,11 @@
 @Builder(toBuilder = true)
 @JsonIgnoreProperties(ignoreUnknown = true)
 @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
-public record FeedbackScoreBatch(@NotNull @Size(min = 1, max = 1000) @Valid List<FeedbackScoreBatchItem> scores) {
+public record FeedbackScoreBatch(
+        @NotNull @Size(min = 1, max = 1000) @Valid List<FeedbackScoreBatchItem> scores) implements RateEventContainer {
 
+    @Override
+    public long eventCount() {
+        return scores.size();
+    }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/SpanBatch.java

@@ -1,5 +1,6 @@
 package com.comet.opik.api;
 
+import com.comet.opik.infrastructure.ratelimit.RateEventContainer;
 import com.fasterxml.jackson.annotation.JsonView;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotNull;
@@ -10,5 +11,10 @@
 
 @Builder(toBuilder = true)
 public record SpanBatch(@NotNull @Size(min = 1, max = 1000) @JsonView( {
-        Span.View.Write.class}) @Valid List<Span> spans){
+        Span.View.Write.class}) @Valid List<Span> spans) implements RateEventContainer{
+
+    @Override
+    public long eventCount() {
+        return spans.size();
+    }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/TraceBatch.java

@@ -1,5 +1,6 @@
 package com.comet.opik.api;
 
+import com.comet.opik.infrastructure.ratelimit.RateEventContainer;
 import com.fasterxml.jackson.annotation.JsonView;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotNull;
@@ -10,5 +11,10 @@
 
 @Builder(toBuilder = true)
 public record TraceBatch(@NotNull @Size(min = 1, max = 1000) @JsonView( {
-        Trace.View.Write.class}) @Valid List<Trace> traces){
+        Trace.View.Write.class}) @Valid List<Trace> traces) implements RateEventContainer{
+
+    @Override
+    public long eventCount() {
+        return traces.size();
+    }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/DatasetsResource.java

@@ -16,6 +16,7 @@
 import com.comet.opik.domain.FeedbackScoreDAO;
 import com.comet.opik.domain.IdGenerator;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.ratelimit.RateLimited;
 import com.comet.opik.utils.AsyncUtils;
 import com.comet.opik.utils.JsonUtils;
 import com.fasterxml.jackson.annotation.JsonView;
@@ -136,6 +137,7 @@ public Response findDatasets(
                     @Header(name = "Location", required = true, example = "${basePath}/api/v1/private/datasets/{id}", schema = @Schema(implementation = String.class))
             })
     })
+    @RateLimited
     public Response createDataset(
             @RequestBody(content = @Content(schema = @Schema(implementation = Dataset.class))) @JsonView(Dataset.View.Write.class) @NotNull @Valid Dataset dataset,
             @Context UriInfo uriInfo) {
@@ -156,6 +158,7 @@ public Response createDataset(
     @Operation(operationId = "updateDataset", summary = "Update dataset by id", description = "Update dataset by id", responses = {
             @ApiResponse(responseCode = "204", description = "No content"),
     })
+    @RateLimited
     public Response updateDataset(@PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = DatasetUpdate.class))) @NotNull @Valid DatasetUpdate datasetUpdate) {
 
@@ -346,6 +349,7 @@ private void sendDatasetItems(DatasetItem item, ChunkedOutput<JsonNode> writer)
     @Operation(operationId = "createOrUpdateDatasetItems", summary = "Create/update dataset items", description = "Create/update dataset items based on dataset item id", responses = {
             @ApiResponse(responseCode = "204", description = "No content"),
     })
+    @RateLimited
     public Response createDatasetItems(
             @RequestBody(content = @Content(schema = @Schema(implementation = DatasetItemBatch.class))) @JsonView({
                     DatasetItem.View.Write.class}) @NotNull @Valid DatasetItemBatch batch) {

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/ExperimentsResource.java

@@ -10,6 +10,7 @@
 import com.comet.opik.domain.ExperimentService;
 import com.comet.opik.domain.IdGenerator;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.ratelimit.RateLimited;
 import com.comet.opik.utils.AsyncUtils;
 import com.fasterxml.jackson.annotation.JsonView;
 import io.dropwizard.jersey.errors.ErrorMessage;
@@ -106,6 +107,7 @@ public Response get(@PathParam("id") UUID id) {
     @Operation(operationId = "createExperiment", summary = "Create experiment", description = "Create experiment", responses = {
             @ApiResponse(responseCode = "201", description = "Created", headers = {
                     @Header(name = "Location", required = true, example = "${basePath}/v1/private/experiments/{id}", schema = @Schema(implementation = String.class))})})
+    @RateLimited
     public Response create(
             @RequestBody(content = @Content(schema = @Schema(implementation = Experiment.class))) @JsonView(Experiment.View.Write.class) @NotNull @Valid Experiment experiment,
             @Context UriInfo uriInfo) {
@@ -151,6 +153,7 @@ public Response getExperimentItem(@PathParam("id") UUID id) {
     @Path("/items")
     @Operation(operationId = "createExperimentItems", summary = "Create experiment items", description = "Create experiment items", responses = {
             @ApiResponse(responseCode = "204", description = "No content")})
+    @RateLimited
     public Response createExperimentItems(
             @RequestBody(content = @Content(schema = @Schema(implementation = ExperimentItemsBatch.class))) @NotNull @Valid ExperimentItemsBatch request) {
 

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/FeedbackDefinitionResource.java

@@ -6,6 +6,7 @@
 import com.comet.opik.api.Page;
 import com.comet.opik.domain.FeedbackDefinitionService;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.ratelimit.RateLimited;
 import com.fasterxml.jackson.annotation.JsonView;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.headers.Header;
@@ -100,6 +101,7 @@ public Response getById(@PathParam("id") @NotNull UUID id) {
             @ApiResponse(responseCode = "201", description = "Created", headers = {
                     @Header(name = "Location", required = true, example = "${basePath}/v1/private/feedback-definitions/{feedbackId}", schema = @Schema(implementation = String.class))})
     })
+    @RateLimited
     public Response create(
             @RequestBody(content = @Content(schema = @Schema(implementation = FeedbackDefinition.class))) @JsonView({
                     FeedbackDefinition.View.Create.class}) @NotNull @Valid FeedbackDefinition<?> feedbackDefinition,
@@ -123,6 +125,7 @@ public Response create(
     @Operation(operationId = "updateFeedbackDefinition", summary = "Update feedback definition by id", description = "Update feedback definition by id", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")
     })
+    @RateLimited
     public Response update(final @PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = FeedbackDefinition.class))) @JsonView({
                     FeedbackDefinition.View.Update.class}) @NotNull @Valid FeedbackDefinition<?> feedbackDefinition) {

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/ProjectsResource.java

@@ -8,6 +8,7 @@
 import com.comet.opik.api.error.ErrorMessage;
 import com.comet.opik.domain.ProjectService;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.ratelimit.RateLimited;
 import com.fasterxml.jackson.annotation.JsonView;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.headers.Header;
@@ -100,6 +101,7 @@ public Response getById(@PathParam("id") UUID id) {
             @ApiResponse(responseCode = "422", description = "Unprocessable Content", content = @Content(schema = @Schema(implementation = ErrorMessage.class))),
             @ApiResponse(responseCode = "400", description = "Bad Request", content = @Content(schema = @Schema(implementation = ErrorMessage.class)))
     })
+    @RateLimited
     public Response create(
             @RequestBody(content = @Content(schema = @Schema(implementation = Project.class))) @JsonView(Project.View.Write.class) @Valid Project project,
             @Context UriInfo uriInfo) {
@@ -125,6 +127,7 @@ public Response create(
             @ApiResponse(responseCode = "422", description = "Unprocessable Content", content = @Content(schema = @Schema(implementation = ErrorMessage.class))),
             @ApiResponse(responseCode = "400", description = "Bad Request", content = @Content(schema = @Schema(implementation = ErrorMessage.class)))
     })
+    @RateLimited
     public Response update(@PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = ProjectUpdate.class))) @Valid ProjectUpdate project) {
 

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/SpansResource.java

@@ -14,6 +14,7 @@
 import com.comet.opik.domain.SpanService;
 import com.comet.opik.domain.SpanType;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.ratelimit.RateLimited;
 import com.comet.opik.utils.AsyncUtils;
 import com.fasterxml.jackson.annotation.JsonView;
 import io.swagger.v3.oas.annotations.Operation;
@@ -126,6 +127,7 @@ public Response getById(@PathParam("id") @NotNull UUID id) {
     @Operation(operationId = "createSpan", summary = "Create span", description = "Create span", responses = {
             @ApiResponse(responseCode = "201", description = "Created", headers = {
                     @Header(name = "Location", required = true, example = "${basePath}/v1/private/spans/{spanId}", schema = @Schema(implementation = String.class))})})
+    @RateLimited
     public Response create(
             @RequestBody(content = @Content(schema = @Schema(implementation = Span.class))) @JsonView(Span.View.Write.class) @NotNull @Valid Span span,
             @Context UriInfo uriInfo) {
@@ -148,6 +150,7 @@ public Response create(
     @Path("/batch")
     @Operation(operationId = "createSpans", summary = "Create spans", description = "Create spans", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
+    @RateLimited
     public Response createSpans(
             @RequestBody(content = @Content(schema = @Schema(implementation = SpanBatch.class))) @JsonView(Span.View.Write.class) @NotNull @Valid SpanBatch spans) {
 
@@ -173,6 +176,7 @@ public Response createSpans(
     @Operation(operationId = "updateSpan", summary = "Update span by id", description = "Update span by id", responses = {
             @ApiResponse(responseCode = "204", description = "No Content"),
             @ApiResponse(responseCode = "404", description = "Not found")})
+    @RateLimited
     public Response update(@PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = SpanUpdate.class))) @NotNull @Valid SpanUpdate spanUpdate) {
 
@@ -201,6 +205,7 @@ public Response deleteById(@PathParam("id") @NotNull String id) {
     @Path("/{id}/feedback-scores")
     @Operation(operationId = "addSpanFeedbackScore", summary = "Add span feedback score", description = "Add span feedback score", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
+    @RateLimited
     public Response addSpanFeedbackScore(@PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = FeedbackScore.class))) @NotNull @Valid FeedbackScore score) {
 
@@ -236,6 +241,7 @@ public Response deleteSpanFeedbackScore(@PathParam("id") UUID id,
     @Path("/feedback-scores")
     @Operation(operationId = "scoreBatchOfSpans", summary = "Batch feedback scoring for spans", description = "Batch feedback scoring for spans", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
+    @RateLimited
     public Response scoreBatchOfSpans(
             @RequestBody(content = @Content(schema = @Schema(implementation = FeedbackScoreBatch.class))) @NotNull @Valid FeedbackScoreBatch batch) {
 

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/TracesResource.java

@@ -15,6 +15,7 @@
 import com.comet.opik.domain.FeedbackScoreService;
 import com.comet.opik.domain.TraceService;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.ratelimit.RateLimited;
 import com.comet.opik.utils.AsyncUtils;
 import com.fasterxml.jackson.annotation.JsonView;
 import io.swagger.v3.oas.annotations.Operation;
@@ -125,6 +126,7 @@ public Response getById(@PathParam("id") UUID id) {
     @Operation(operationId = "createTrace", summary = "Create trace", description = "Get trace", responses = {
             @ApiResponse(responseCode = "201", description = "Created", headers = {
                     @Header(name = "Location", required = true, example = "${basePath}/v1/private/traces/{traceId}", schema = @Schema(implementation = String.class))})})
+    @RateLimited
     public Response create(
             @RequestBody(content = @Content(schema = @Schema(implementation = Trace.class))) @JsonView(Trace.View.Write.class) @NotNull @Valid Trace trace,
             @Context UriInfo uriInfo) {
@@ -150,7 +152,8 @@ public Response create(
     @Path("/batch")
     @Operation(operationId = "createTraces", summary = "Create traces", description = "Create traces", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
-    public Response createSpans(
+    @RateLimited
+    public Response createTraces(
             @RequestBody(content = @Content(schema = @Schema(implementation = TraceBatch.class))) @JsonView(Trace.View.Write.class) @NotNull @Valid TraceBatch traces) {
 
         traces.traces()
@@ -174,6 +177,7 @@ public Response createSpans(
     @Path("{id}")
     @Operation(operationId = "updateTrace", summary = "Update trace by id", description = "Update trace by id", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
+    @RateLimited
     public Response update(@PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = TraceUpdate.class))) @Valid @NonNull TraceUpdate trace) {
 
@@ -225,6 +229,7 @@ public Response deleteTraces(
     @Path("/{id}/feedback-scores")
     @Operation(operationId = "addTraceFeedbackScore", summary = "Add trace feedback score", description = "Add trace feedback score", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
+    @RateLimited
     public Response addTraceFeedbackScore(@PathParam("id") UUID id,
             @RequestBody(content = @Content(schema = @Schema(implementation = FeedbackScore.class))) @NotNull @Valid FeedbackScore score) {
 
@@ -265,6 +270,7 @@ public Response deleteTraceFeedbackScore(@PathParam("id") UUID id,
     @Path("/feedback-scores")
     @Operation(operationId = "scoreBatchOfTraces", summary = "Batch feedback scoring for traces", description = "Batch feedback scoring for traces", responses = {
             @ApiResponse(responseCode = "204", description = "No Content")})
+    @RateLimited
     public Response scoreBatchOfTraces(
             @RequestBody(content = @Content(schema = @Schema(implementation = FeedbackScoreBatch.class))) @NotNull @Valid FeedbackScoreBatch batch) {
 

apps/opik-backend/src/main/java/com/comet/opik/domain/FeedbackScoreService.java

@@ -8,7 +8,7 @@
 import com.comet.opik.api.error.IdentifierMismatchException;
 import com.comet.opik.infrastructure.auth.RequestContext;
 import com.comet.opik.infrastructure.db.TransactionTemplate;
-import com.comet.opik.infrastructure.redis.LockService;
+import com.comet.opik.infrastructure.lock.LockService;
 import com.comet.opik.utils.WorkspaceUtils;
 import com.google.inject.ImplementedBy;
 import com.google.inject.Singleton;

apps/opik-backend/src/main/java/com/comet/opik/domain/SpanService.java

@@ -10,7 +10,7 @@
 import com.comet.opik.api.error.ErrorMessage;
 import com.comet.opik.api.error.IdentifierMismatchException;
 import com.comet.opik.infrastructure.auth.RequestContext;
-import com.comet.opik.infrastructure.redis.LockService;
+import com.comet.opik.infrastructure.lock.LockService;
 import com.comet.opik.utils.WorkspaceUtils;
 import com.google.common.base.Preconditions;
 import com.newrelic.api.agent.Trace;

apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java

@@ -12,7 +12,7 @@
 import com.comet.opik.api.error.IdentifierMismatchException;
 import com.comet.opik.infrastructure.auth.RequestContext;
 import com.comet.opik.infrastructure.db.TransactionTemplate;
-import com.comet.opik.infrastructure.redis.LockService;
+import com.comet.opik.infrastructure.lock.LockService;
 import com.comet.opik.utils.AsyncUtils;
 import com.comet.opik.utils.WorkspaceUtils;
 import com.google.common.base.Preconditions;

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/OpikConfiguration.java

@@ -33,4 +33,8 @@ public class OpikConfiguration extends Configuration {
     @Valid
     @NotNull @JsonProperty
     private DistributedLockConfig distributedLock = new DistributedLockConfig();
+
+    @Valid
+    @NotNull @JsonProperty
+    private RateLimitConfig rateLimit = new RateLimitConfig();
 }