feat(stream-service-core): shared-cluster Fleet CDC consumption (per-event tenant) - #1572
Merged
Merged
Conversation
…event tenant) Under Fleet shared-DB multi-tenancy one Debezium connector captures every tenant's Fleet MySQL into cluster-neutral raw topics, consumed by the SHARED openframe-saas-stream (the MeshCentral pattern). This adds the library side of that per-event tenant resolution: - ClusterTenantIdResolver: tool-aware default method resolveTenantId(IntegratedToolType, key) — backward compatible (defaults to the historical single-arg lookup); enrichment now dispatches with the event's tool type (MeshCentral domain vs Fleet team_id). - Fleet deserializers surface the fork-stamped team_id via the getTenantId hook; Activity maps team_id so it survives the activity/host-activity Kafka Streams join round-trip. - FleetMdmCacheService: per-event-tenant clients (shared credential, per-tenant X-Tenant-Id — the shared Fleet's fences 404 lookups made under the wrong tenant). Cache keys stay id-only (ids are globally unique on the shared DB). Startup validation is skipped when a ClusterTenantIdResolver is present (shared cluster: no deployment TENANT_ID by design); an unresolved event tenant skips the lookup instead of making a doomed header-less call. - DebeziumCassandraMessageHandler now applies the same tenant guard as the Kafka/Pinot handler: an event with no resolved tenant is dropped instead of failing the write on the NULL tenant_id clustering column. Deliberately no isVisible check — invisible events stay in the event log. Inert in tenant clusters (DefaultTenantIdProvider never returns blank). - KafkaStreamsConfig: openframe.stream.kafka-streams.bootstrap-servers override so the activity join can run against a specific cluster when both Kafka configs are present (shared deployment); unset keeps the old fallback chain byte-for-byte. Backward compatible: no downstream constructions/subclasses of the changed classes exist; tenant-cluster behavior verified unchanged (env tenant overwrites the raw team_id, same cache names/keys, guard never trips); downstream stream services compile clean against this snapshot. 65/65 module tests green (3 new test classes). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Warning Review limit reached
Next review available in: 43 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughFleet CDC events now extract Fleet ChangesFleet shared-cluster tenancy
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant FleetCDC
participant FleetDeserializer
participant ClusterTenantIdResolver
participant FleetMdmCacheService
participant FleetMdmClient
FleetCDC->>FleetDeserializer: emit event with team_id
FleetDeserializer->>ClusterTenantIdResolver: resolve Fleet tenant
ClusterTenantIdResolver-->>FleetDeserializer: canonical tenant ID
FleetDeserializer->>FleetMdmCacheService: request policy/query with event tenant
FleetMdmCacheService->>FleetMdmClient: perform tenant-scoped lookup
FleetMdmClient-->>FleetMdmCacheService: return metadata
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
openframe-stream-service-core/src/main/java/com/openframe/stream/service/FleetMdmCacheService.java (1)
184-238: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftBound the per-tenant Fleet client cache and refresh cached credentials.
clientByTenantis a plainConcurrentHashMap, so each distinct resolvedeventTenantIdpermanently keeps a newFleetMdmClientwith its ownHttpClient; in shared-cluster deployments this can grow without bound.cachedApiKeyis similarly retained forever with no refresh path if the Fleet API key rotates. Use a bounded TTL-based cache such as Caffeine for tenants and add an invalidation/refresh hook for credential changes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openframe-stream-service-core/src/main/java/com/openframe/stream/service/FleetMdmCacheService.java` around lines 184 - 238, Update clientByTenant in clientFor to use a bounded, TTL-based cache such as Caffeine so resolved tenant IDs cannot retain FleetMdmClient instances indefinitely; configure both an entry limit and expiration policy while preserving existing lazy client creation. Update resolveApiKey and credential-change handling to invalidate cachedApiKey and the tenant client cache when Fleet credentials rotate, allowing subsequent lookups to load the new key and recreate clients.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@openframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/FleetQueryResultEventDeserializer.java`:
- Line 211: Partition Fleet metadata cache keys by tenant: update
FleetMdmCacheService.getQueryById(Long, String) to include eventTenantId, and
update FleetMdmCacheService.getPolicyById(Long, String) likewise. Apply the
corresponding tenant-scoped changes at
openframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/FleetQueryResultEventDeserializer.java:211-211
and
openframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/FleetPolicyMembershipEventDeserializer.java:152-152.
In
`@openframe-stream-service-core/src/main/java/com/openframe/stream/handler/DebeziumCassandraMessageHandler.java`:
- Around line 45-54: Update DebeziumCassandraMessageHandler.isValidMessage to
require both Boolean.TRUE.equals(message.getIsVisible()) and
eventValidator.isValid(message), matching the existing Kafka/Pinot visibility
gate while retaining tenant validation.
---
Nitpick comments:
In
`@openframe-stream-service-core/src/main/java/com/openframe/stream/service/FleetMdmCacheService.java`:
- Around line 184-238: Update clientByTenant in clientFor to use a bounded,
TTL-based cache such as Caffeine so resolved tenant IDs cannot retain
FleetMdmClient instances indefinitely; configure both an entry limit and
expiration policy while preserving existing lazy client creation. Update
resolveApiKey and credential-change handling to invalidate cachedApiKey and the
tenant client cache when Fleet credentials rotate, allowing subsequent lookups
to load the new key and recreate clients.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: deef7f42-0a0d-4760-833d-14e49e133657
📒 Files selected for processing (13)
openframe-stream-service-core/src/main/java/com/openframe/stream/config/KafkaStreamsConfig.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/FleetEventDeserializer.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/FleetPolicyMembershipEventDeserializer.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/FleetQueryResultEventDeserializer.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/deserializer/IntegratedToolEventDeserializer.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/handler/DebeziumCassandraMessageHandler.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/model/fleet/Activity.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/service/ClusterTenantIdResolver.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/service/FleetMdmCacheService.javaopenframe-stream-service-core/src/main/java/com/openframe/stream/service/IntegratedToolDataEnrichmentService.javaopenframe-stream-service-core/src/test/java/com/openframe/stream/deserializer/FleetCdcTeamTenantTest.javaopenframe-stream-service-core/src/test/java/com/openframe/stream/handler/DebeziumCassandraMessageHandlerTenantGuardTest.javaopenframe-stream-service-core/src/test/java/com/openframe/stream/service/FleetMdmCacheServiceSharedClusterTest.java
…ivity tenant threading Addresses CodeRabbit review on #1572: - Tenant-aware cache variants now include the event tenant in the cache key ((#eventTenantId ?: 'default') + ':' + id). Ids are globally unique on today's single shared MySQL, so no cross-tenant collision exists — but tenant-scoped keys stay correct if the DB ever shards per cluster (ids unique per shard only) and on dev where several SHARED_DOMAIN environments share one Redis. Matching tenant-aware evictPolicyCache added. - Review follow-up surfaced a real gap: FleetPolicyActivityDeserializer (policy-CRUD activities, FLEET_MDM_POLICY_ACTIVITY_EVENT) was never tenant-threaded — in shared mode its events would have been dropped as tenant-unresolved (no getTenantId override) and its policy lookups made header-less. It now surfaces the stamped team_id, resolves the event tenant, and uses the tenant-aware lookup + evict like the other Fleet deserializers. 65/65 module tests green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
aliaska-varieva
approved these changes
Jul 27, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Under Fleet shared-DB multi-tenancy, one Debezium connector captures every tenant's Fleet MySQL into cluster-neutral raw topics consumed by the SHARED
openframe-saas-stream— the proven MeshCentral pattern (one shared upstream, per-event tenant resolution, fail-closed drop, tenant-keyed shared sinks). This PR adds the library side of that consumption; the shared listener/teams-bridge cache and the connector manifests land inopenframe-saas-shared.Ticket: https://app.clickup.com/t/9013925967/86ajmwa0c
Changes
ClusterTenantIdResolver— new tool-awaredefault resolveTenantId(IntegratedToolType, key). The tenant discriminator arrives in the same field for every tool but means different things (MeshCentraldomainvs Fleetteam_id), so resolution dispatches on the event's tool type. The default delegates to the historical single-arg lookup — existing implementers/callers (RmmEnrichmentService, MeshCentral flow) are untouched.IntegratedToolDataEnrichmentServicenow calls the tool-aware form.team_idvia thegetTenantIdhook (shared helperextractFleetTeamId);Activitymapsteam_idso it survives the activity/host-activity Kafka Streams join's deserialize/re-serialize round-trip.FleetMdmCacheService— per-event-tenant clients: one shared credential, per-tenantX-Tenant-Id(the shared Fleet's fences 404 a query/policy lookup made under the wrong tenant). Cache keys stay id-only — host/query/policy ids are globally unique on the shared DB. Startup validation is skipped when aClusterTenantIdResolverbean is present (shared cluster: no deploymentTENANT_IDby design), and an unresolved event tenant skips the Fleet lookup instead of making a doomed header-less call.DebeziumCassandraMessageHandler— now applies the same tenant guard as the Kafka/Pinot handler: an event whose tenant could not be resolved (e.g. a Fleet CDC row without a stampedteam_id, or an unmapped team) is dropped fail-closed instead of failing the write on the NULLtenant_idclustering column ofunified_logs. Deliberately noisVisiblecheck — invisible events remain stored in the event log, only excluded from the Pinot feed.KafkaStreamsConfig—openframe.stream.kafka-streams.bootstrap-serversoverride, first in the fallback chain, so the shared deployment can run the activity join against the SAAS Kafka whilespring.oss-tenant.kafkais also configured. Unset ⇒ old chain, byte-for-byte.Backward compatibility
FleetMdmCacheService,DebeziumCassandraMessageHandler, Fleet deserializers) verified to have zero direct constructions or subclasses inopenframe-saas-lib/openframe-saas-shared/openframe-saas-tenant— all Spring-wired; the added resolver params are@Autowired(required = false).team_idduring enrichment, cache names/keys are identical, and the new Cassandra guard can never trip there (DefaultTenantIdProvidernever returns blank — defaults to"oss").team_idfield; consumers map the payload asJsonNodeand ignore it.openframe-saas-tenantandopenframe-saas-sharedstream services compile clean against this snapshot; saas-shared stream tests 9/9.Tests
65/65 in
openframe-stream-service-core, including 3 new classes:FleetCdcTeamTenantTest(team_id surfaced by all Fleet deserializers; enrichment lookups carry the event's resolved tenant; tenant-cluster fallback passes null),DebeziumCassandraMessageHandlerTenantGuardTest(drop on null/blank tenant, write on resolved tenant),FleetMdmCacheServiceSharedClusterTest(unresolved tenant in shared cluster skips the lookup without touching credentials; tenant cluster keeps the deployment-client fallback).Depends on / release order
Pairs with fleetmdm #77 (the
team_idstamping this consumes). This library must release before or together with theopenframe-saas-sharedchange whoseSharedClusterTenantIdResolveroverrides the new tool-aware method; the reverse direction (new lib, old saas-shared) is safe via the default method.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes