Repository navigation
fix(aws-kafka): async lifecycle, scheduled AWS/Kafka metrics, real broker subnets, managed KMS key - #1654
Conversation
NitinKumar004
left a comment
NitinKumar004
left a comment
There was a problem hiding this comment.
What I checked
- All six items in #1517 against the PR head (ea11d32): broker client subnets, CREATING, DELETING, VPC connection CREATING, AWS/Kafka metrics with SetMonitoring wiring, and the managed at-rest key. Each one is handled in the provider. The wire layer only lost its hard-coded
key/msk-defaultblock and now surfaces the stored encryptionInfo for v1 and v2. - Layering looks right. Lifecycle, metrics and encryption defaults live in
providers/aws/kafka, the factory wiresKafka.SetMonitoring(CloudWatch), and Kafka ticks before CloudWatch inTickables(). No driver change, andgo generate .leaves no diff. - Gates on the touched packages:
go build ./...ok,go vetok, scopedgo testok (kafka provider, kafka wire, providers/aws, persist),-raceok, gofmt clean on the touched files,go mod tidyclean. I also ran a throwaway concurrent create/describe/list/delete/Tick test on one cluster name under async settle with-race. It was clean and left no clusters behind. - golangci-lint (v2.14.0) on
./providers/aws/kafka/... ./server/aws/kafka/... ./providers/aws/without--new-from-rev, compared with the merge base: one new issue (see Medium), plusnewProviderfunlen going from 66 to 67 statements. That funlen hit was already failing on the base. cloudemu serve -async-settlewith AWS CLI v2:- create-cluster returns CREATING.
- update-broker-count while CREATING returns BadRequestException, and so does an update while DELETING.
- The cluster becomes ACTIVE after about 3s.
- list-nodes gives subnet-0aaa/0bbb/0ccc.
- The v1 and v2 describes report the same managed key ARN with TLS/true.
- delete returns DELETING, and a second delete also returns DELETING.
- Creating a cluster with the same name while DELETING returns ConflictException.
- After the window, describe and list-nodes return NotFoundException.
- create-vpc-connection shows CREATING, then AVAILABLE. An unknown ARN returns NotFoundException.
- After about a minute, list-metrics shows the 4 cluster-level metrics and the 2 per-broker metrics. get-metric-statistics returns one datapoint per minute with SampleCount 1, Maximum 1 and Unit Count.
- Default
serve(no async): ACTIVE immediately, 4 brokers over 2 subnets gives a/b/a/b, delete then describe returns NotFound, the name can be reused right away, and serverless v2 create still works. - Terraform hashicorp/aws 6.66.0, two
aws_msk_clusterresources (one with the default key, one with an explicitencryption_at_rest_kms_key_arn), againstserve -async-settle:- apply took about 3s per cluster, and the create waiter observed CREATING then ACTIVE.
- plan exited 0.
- Updating the broker count from 3 to 6 and changing tags applied cleanly, and the next plan exited 0.
- destroy finished, and describe then returned NotFoundException.
- The state for the default cluster carries the managed key ARN with
client_broker = "TLS"andin_cluster = true.
- The PR body and commit message have no attribution lines.
High
None.
Medium
(*Mock).Snapshotinproviders/aws/kafka/snapshot.go:61now has a cyclomatic complexity of 12, over the limit of 10 (gocyclo). The function was at the limit before this change, and the newvpcConnReadyAtbranch (if !vd.readyAt.IsZero()plus the lazy map init) pushed it over. A--new-from-revrun hides this because gocyclo reports on thefuncline, which this diff does not touch. That is why the PR's "0 issues" claim does not hold for a full lint of the package. Moving the VPC connection loop into a small helper such assnapshotVPCConns(snap *kafkaSnapshot)fixes it and keeps the function readable.
Low
-
withDefaultEncryption(providers/aws/kafka/encryption.go:58-85) fills in only the at-rest key when the caller sends a partialencryptionInfo. Real MSK also reports the in-transit defaults in that case:--encryption-info '{"EncryptionAtRest":{"DataVolumeKMSKeyId":"...mine"}}'describes with noEncryptionInTransitat all. Real MSK reportsClientBroker=TLS, InCluster=true.{"EncryptionInTransit":{"ClientBroker":"TLS_PLAINTEXT"}}describes withoutInCluster. Real MSK reportstrue.
Terraform happens not to drift on this, but SDK clients that read
EncryptionInTransit.InClusterget nil. Since this function is the place that "completes EncryptionInfo the way MSK does", filling in the missing in-transit fields there would close the gap. -
Tick(now)uses the scheduler'snowfor the sample cadence and timestamps, butdueMetricsdecides ACTIVE throughm.clusterStateLocked, which readsm.now()(providers/aws/kafka/metrics.go:99). Underserveboth are the wall clock, but a caller that passes anowdifferent from the options clock gets a mix of the two. Passingnowthrough to the state check would make Tick consistent with its argument. -
requireActiveguards only themutateClusterErrfamily. Topic create/update/delete, SCRAM secret association and cluster policy writes still succeed on a CREATING or DELETING cluster. Real MSK rejects API operations until the cluster is ACTIVE. This only matters under--async-settle, so it could be a follow-up. -
GetCompatibleKafkaVersions(providers/aws/kafka/versions.go:95) readsm.clustersdirectly and skipsreapIfGone, so a cluster whose DELETING window has passed still answers until some other read removes it. This is a small edge case. -
reapNamehas 60% coverage. Neither of its two branches is tested: the one where the name points at an ARN that is no longer in the store, and the one where the name is reaped on create without any earlier read. The kafka package overall is at 88.4%, under the 90% target.
| snap.VPCConnReadyAt = map[string]time.Time{} | ||
| } | ||
|
|
||
| snap.VPCConnReadyAt[arn] = vd.readyAt |
There was a problem hiding this comment.
This branch pushes (*Mock).Snapshot to a cyclomatic complexity of 12, over the gocyclo limit of 10. A full golangci-lint run ./providers/aws/kafka/... reports it on the func line, which is why --new-from-rev misses it. Could the VPC connection loop move into a small helper like snapshotVPCConns(snap *kafkaSnapshot)?
There was a problem hiding this comment.
Fixed in c86fc4e. Snapshot now delegates to per-store helpers (clustersForSnapshot, configsForSnapshot, vpcConnsForSnapshot, replicatorsForSnapshot), and a full golangci-lint run ./providers/aws/kafka/... ./server/aws/kafka/... (without --new-from-rev) reports 0 issues. While doing this I found a data race that was already there: topics, operations, scramSecrets and the raw-option maps were shared, and json.Marshal read them after the lock was released. Each record is now deep-copied under its own lock. TestSnapshotConcurrentWithMutations reported a race under -race before the fix and is clean after it.
| _ = json.Unmarshal(k, &existing) | ||
| } | ||
|
|
||
| if existing != "" { |
There was a problem hiding this comment.
When the caller sends a partial block, only the at-rest key gets filled in. Real MSK also reports the in-transit defaults here: with EncryptionAtRest only, describe should still show EncryptionInTransit {ClientBroker: TLS, InCluster: true}, and with {ClientBroker: TLS_PLAINTEXT} only, InCluster should come back true. Right now the first case has no EncryptionInTransit at all and the second has no InCluster.
There was a problem hiding this comment.
Fixed in c86fc4e. withDefaultEncryption now fills each missing field on its own with its documented default: the managed at-rest key, clientBroker TLS and inCluster true (EncryptionInTransit). Explicit values, including inCluster: false, are kept. TestDefaultEncryptionInTransitCompleted covers at-rest only, ClientBroker only, InCluster=false only, and {}. I also checked it with the CLI: EncryptionAtRest alone now describes with TLS/true, and TLS_PLAINTEXT alone describes with InCluster: true.
| defer cd.mu.Unlock() | ||
|
|
||
| if cd.cluster.ClusterType != driver.ClusterTypeProvisioned || | ||
| m.clusterStateLocked(cd) != driver.ClusterStateActive { |
There was a problem hiding this comment.
Small consistency point: the cadence and timestamps use Tick's now, but this ACTIVE check goes through clusterStateLocked, which reads m.now(). Passing now through would keep Tick consistent with its argument when the two clocks differ.
There was a problem hiding this comment.
Fixed in c86fc4e. clusterStateAt(cd, t) and goneAt(cd, t) now take the time explicitly. Tick passes its own now for both the ACTIVE check and the new delete reap, and request paths pass m.now(). Test: TestTickUsesItsNow.
|
Thanks for the thorough review, @NitinKumar004. I checked each finding against the code before changing anything. All of them are addressed in c86fc4e. New tests are in Medium1. While I was in there, I found a real data race that was there before this PR. Snapshot copied the cluster struct under Low1. Partial 2. Tick mixing its 3. 4. 5. Package coverage: the provider tests alone give 84.9%, and the provider plus wire tests ( One more change related to this reviewRemoval after a delete used to happen only on the next read. Tick now also removes any cluster whose DELETING window has ended and returns Verification
Follow-up (not in this PR)
|
NitinKumar004
left a comment
NitinKumar004
left a comment
There was a problem hiding this comment.
What I checked
- The new commit c86fc4e against each point from my earlier review, then the whole diff again from the merge base. The PR still merges cleanly into
developmentand every CI check on c86fc4e is green. - Gates on the new head:
go build ./...ok.go vetok on./providers/aws/kafka/... ./server/aws/kafka/... ./providers/aws/. Scopedgo testand-raceok on those packages plus./persist/.... gofmt is clean on the touched files,go mod tidyleaves no diff, andgo generate .leaves no diff. - golangci-lint v2.14.0, full run (no
--new-from-rev) on./providers/aws/kafka/... ./server/aws/kafka/... ./providers/aws/, compared with a merge-base worktree: 53 issues on both sides, and the lists are the same apart from line numbers. The only changed entry isnewProviderfunlen going from 66 to 67 statements, which already fails on the base. The gocyclo hit on(*Mock).Snapshotis gone. - Coverage of
providers/aws/kafka: 89.8% (merge base 86.8%, earlier head 88.4%).reapNameis now at 100%. cloudemu serve -async-settlewith AWS CLI 2.31.19:- create-cluster returns CREATING. UpdateBrokerCount, PutClusterPolicy and BatchAssociateScramSecret while CREATING all return BadRequestException ("is in CREATING state; this operation requires ACTIVE"). GetCompatibleKafkaVersions is still allowed.
- After about 3s the cluster is ACTIVE. The policy put and the SCRAM associate then succeed. list-nodes gives subnet-0aaa/0bbb/0ccc.
- Partial encryptionInfo: at-rest key only describes with
ClientBroker=TLS, InCluster=true.ClientBroker=TLS_PLAINTEXTonly describes withInCluster=trueand the managed key.InCluster=falseonly is kept as false, withClientBroker=TLS. describe-cluster-v2 shows the same block. - delete-cluster returns DELETING. PutClusterPolicy and BatchDisassociateScramSecret while DELETING return BadRequestException. Creating a cluster with the same name returns ConflictException.
- After the window, list-clusters-v2 no longer shows the cluster. GetCompatibleKafkaVersions and describe-cluster return NotFoundException, and the name can be reused.
- Default
serve(no async): create returns ACTIVE, writes work right away, delete then describe returns NotFoundException, and the name can be reused at once. - Terraform 1.14.3 with hashicorp/aws 6.70.0 (downloaded into a private directory), two
aws_msk_clusterresources, one with noencryption_infoand one with onlyencryption_in_transit { client_broker = "TLS_PLAINTEXT" }, againstserve -async-settle:- apply: each cluster was created in about 3s.
- plan
-detailed-exitcodereturned 0. The state for the partial cluster has the managed key ARN,TLS_PLAINTEXTandin_cluster = true. - I changed the broker count from 3 to 6 and changed a tag. apply worked and plan returned 0 again.
- list-metrics showed 10 AWS/Kafka metrics for the cluster with no encryption block: 4 for the cluster and 2 for each of its 3 brokers.
- destroy took about 3s per cluster, and describe then returned NotFoundException.
- The PR body and commits have no attribution lines.
Status of the earlier points
- Medium 1, gocyclo on
(*Mock).Snapshot: fixed. Snapshot now callsclustersForSnapshot,configsForSnapshot,vpcConnsForSnapshotandreplicatorsForSnapshot. A full lint run no longer reports it. Each record is also deep-copied under its own lock now. That fixes the existing race wherejson.Marshalread topic, operation, SCRAM and raw-option maps after the lock was released.TestSnapshotConcurrentWithMutationspasses under-race. - Low 1, partial encryptionInfo: fixed.
defaultFieldfills each missing field on its own, and an explicitinCluster: falseis kept. I confirmed this with the CLI and Terraform as described above. - Low 2, Tick mixing clocks: fixed. Tick passes its
nowto bothclusterStateAtandreapIfGoneAt. Request paths still usem.now().TestTickUsesItsNowcovers it. - Low 3, topic, SCRAM and policy writes not guarded: fixed.
requireActivenow runs undercd.muin CreateTopic, UpdateTopic, DeleteTopic, BatchAssociateScramSecret, BatchDisassociateScramSecret, PutClusterPolicy and DeleteClusterPolicy. I checked the SCRAM and policy cases live. - Low 4, GetCompatibleKafkaVersions skipping the reap: fixed.
compatibilitySourcesnow goes throughgetCluster, so the reap runs and a cluster past its window returns NotFoundException.TestCompatibleVersionsAfterDeleteWindowcovers it. - Low 5, coverage: mostly fixed.
reapNameis at 100% (TestCreateReclaimsDanglingName). The package is at 89.8%, just under the 90% target (see Low 1 below).
Tick now also removes clusters whose delete window has ended. The lock order is fine: reapIfGoneAt releases cd.mu before it takes reapMu, and Tick holds neither createMu nor any cluster lock at that point, so it follows createMu -> reapMu -> clusterData.mu.
High
None.
Medium
None.
Low
- The kafka provider package is at 89.8%, 0.2 points under the 90% target. The largest gap in the touched code is
compatibilitySourcesinproviders/aws/kafka/versions.goat 55.6%. Its no-cluster-ARN branch and the unknown-ARNgetClustererror branch are not tested. One test that calls GetCompatibleKafkaVersions with an empty ARN and one with an unknown ARN would get the package over 90%.
| cd, ok := m.clusters.Get(clusterARN) | ||
| if !ok { | ||
| cd, err := m.getCluster(clusterARN) | ||
| if err != nil { |
There was a problem hiding this comment.
Small coverage gap: compatibilitySources is at 55.6%, and this error branch (unknown ARN) and the empty-ARN branch above are not tested. Tests for those two cases would get the package from 89.8% to over 90%.
…oker subnets, managed KMS key Close the open real-cloud parity gaps in stackshy#1517: - ListNodes reports each broker's clientSubnet from the cluster's own BrokerNodeGroupInfo.ClientSubnets, spreading brokers evenly across them (round-robin by broker ID), instead of a fabricated subnet-<brokerId>. The field is omitted when the cluster records no subnets. - Under AsyncSettle a new cluster reports CREATING (create response, Describe, List) and settles to ACTIVE; a deleted cluster reports DELETING and is gone (NotFoundException) once its window ends, keeping its name until then. Update operations on a non-ACTIVE cluster are rejected with BadRequestException. A new VPC connection reports CREATING, then AVAILABLE. The default (AsyncSettle off) stays synchronous. The windows are per-record deadlines persisted in snapshots. - Publish DEFAULT-level AWS/Kafka metrics from the shared serve scheduler (Kafka is now a config.Tickable, ticked before CloudWatch), once a minute per ACTIVE provisioned cluster: ActiveControllerCount, GlobalTopicCount, GlobalPartitionCount and OfflinePartitionsCount on "Cluster Name", and UnderReplicatedPartitions and UnderMinIsrPartitionCount on "Cluster Name" + "Broker ID". Host and traffic metrics cloudemu does not model are not published. Kafka.SetMonitoring is wired to CloudWatch. - The default encryption-at-rest key is the account+region's AWS managed aws/kafka key ARN (one stable UUID key id per account and region), completed by the provider at create time and reported by both v1 and v2 describes, instead of a hard-coded key/msk-default in us-east-1. Part of stackshy#1517
…ncryption defaults - Snapshot deep-copies each cluster, configuration, VPC connection and replicator under its own lock (topics, operations, SCRAM secrets, tags, raw options), so encoding no longer races a concurrent CreateTopic. The per-store loops move into helpers, which also brings Snapshot back under the gocyclo limit. - withDefaultEncryption fills every EncryptionInfo field the request leaves out with its documented default (managed at-rest key, ClientBroker TLS, InCluster true) and keeps explicit values, including InCluster false. - Tick judges cluster state at its own now argument instead of the options clock, and removes clusters whose DELETING window has ended, reporting the change so persist saves it even when nothing reads the cluster. - Topic create/update/delete, SCRAM secret associate/disassociate and cluster policy put/delete are rejected with BadRequestException while a cluster is CREATING or DELETING (only reachable under --async-settle). - GetCompatibleKafkaVersions resolves the cluster through getCluster, so a cluster past its DELETING window is NotFound. Part of stackshy#1517
c86fc4e to
d902f07
Compare
Summary
--async-settle. The default mode is still synchronous.Item status
clientSubnetassubnet-<brokerId>providers/aws/kafka/nodes.go. The API reference says: "When you create a cluster, Amazon MSK distributes the broker nodes evenly across the subnets that you specify" (BrokerNodeGroupInfo). Brokers now get the cluster's ownClientSubnetsin round-robin order by broker ID. If the cluster has no subnets,clientSubnetis left out.--async-settle, CreateCluster/CreateClusterV2, Describe and List showCREATINGforsettle.DefaultClusterSettle(3s), thenACTIVE. Update operations on a cluster that is not ACTIVE returnBadRequestException. The MSK docs say that in CREATING you "must wait for the cluster to reach the ACTIVE state before you can ... perform Amazon MSK API or AWS CLI operations on it" (cluster states).Kafka.SetMonitoring(CloudWatch)is wired. Kafka is now aconfig.Tickableand runs before CloudWatch inProvider.Tickables(), so alarms see the metrics from the same tick. MSK pushes metrics "at 1 minute intervals", and the DEFAULT metrics become visible "After the cluster gets to the ACTIVE state" (metrics-details). See the metrics section below.--async-settle, the cluster stays visible asDELETINGin Describe and List and keeps its name, so a create with the same name getsConflictException. After the window it returnsNotFoundException(404). A second delete while DELETING returnsDELETING.VpcConnectionStateisCREATING,AVAILABLE,INACTIVE,UPDATING,DEACTIVATING,DELETING,FAILED,REJECTEDorREJECTING(API reference). With--async-settle, a new connection showsCREATING, thenAVAILABLE.REJECTEDis never hidden.key/msk-defaultEncryptionAtRest, Amazon MSK still encrypts your data at rest under the AWS managed key" (alias/aws/kafka), and DescribeCluster reports that key (encryption). The provider now fills inencryptionAtRest.dataVolumeKMSKeyIdat create time. The value is aarn:aws:kms:<region>:<account>:key/<uuid>ARN that is stable for each account and region, so every cluster in that account and region reports the same key. The in-transit defaults areClientBroker=TLSandInCluster=true, as documented. The wire layer no longer hard-codes aus-east-1:000000000000key.Changes
providers/aws/kafka/nodes.go: brokerclientSubnetcomes from the cluster'sClientSubnets.providers/aws/kafka/lifecycle.go(new): per-record lifecycle deadlines: the cluster'sreadyAt/deleteAtand the VPC connection'sreadyAt.opts.SettleDuration(settle.DefaultClusterSettle), so they are zero, and inactive, unless--async-settleis set.--persistsaves it, or on the next read or create, whichever comes first, underreapMu.requireActiveguardsmutateClusterErr, plus topic create/update/delete, SCRAM secret associate/disassociate and cluster policy put/delete. Reads stay allowed.internal/settle.providers/aws/kafka/metrics.go(new):Tick(now)publishes DEFAULT-level AWS/Kafka metrics once a minute for each ACTIVE provisioned cluster, after all cluster locks are released. It also removes clusters whose DELETING window has ended. Every state check uses Tick'snow.providers/aws/kafka/encryption.go(new): fills in EncryptionInfo the way MSK does. Each field the request leaves out gets its documented default: the managed at-rest key,clientBrokerTLS andinClustertrue. Explicit values, includinginCluster: false, are kept.providers/aws/kafka/snapshot.go: the lifecycle deadlines and the last metric sample time are persisted. Every record is deep-copied under its own lock, which fixes a race with concurrent topic and SCRAM writes that existed before this PR. The per-store loops are now helpers.vpcConnReadyAtsits besidevpcConns, so older snapshots still load.server/aws/kafka/types.go: the hard-codeddefaultEncryptionInfois removed. v1 and v2 describes both show the encryptionInfo stored by the provider. This file only translates the wire format.providers/aws/aws.go: wiresKafka.SetMonitoring(CloudWatch)and addsp.KafkatoTickables().AWS/Kafka metrics published (all
Count, every 60s, ACTIVE provisioned clusters only)These metrics are deliberately not published:
PartitionCountandLeaderCount, because they need a partition assignment that is not modeled.Provider Coverage
Checklist
go test -p 2 -count=1over all 657 packages, run in chunks. Everything passed.golangci-lint run --allow-parallel-runners --concurrency 2 --new-from-rev=upstream/development ./providers/aws/kafka/... ./server/aws/kafka/... ./providers/aws/reports0 issues., and a full run (without--new-from-rev) on./providers/aws/kafka/... ./server/aws/kafka/...also reports0 issues.cloudemu_test.go(not added: the tests are provider tests plus aws-sdk-go-v2 wire tests)go generate .produced nodocs/coveragediff, because no driver interface or wiring of operations changedTest Plan
New tests. Each one fails on
upstream/development, checked with a no-opSetMonitoring/Tickstub so the tests compile, and passes with this change:providers/aws/kafka/parity_test.go(FakeClock):TestListNodesClientSubnetFromClusterTestListNodesNoSubnetsOmitsClientSubnetTestClusterCreatingThenActiveTestDeleteClusterDeletingThenGoneTestVpcConnectionCreatingThenAvailableTestClusterMetricsPublishedByTick: checks that create publishes nothing, nothing is published while CREATING, publishing happens once per minute, and no host metrics are made upTestDefaultEncryptionAtRestKeyTestSnapshotSettleStates: CREATING, DELETING and VPC CREATING all survive snapshot and restore, then settleTestTickSkipsDeletingAndServerless: new; passes on the fixed branch.providers/aws/kafka/review_test.go(review follow-up; each one fails before c86fc4e):TestSnapshotConcurrentWithMutations:-raceTestDefaultEncryptionInTransitCompletedTestTickUsesItsNowTestTickReapsFinishedDeleteTestClusterWritesRequireActiveTestCompatibleVersionsAfterDeleteWindowTestCreateReclaimsDanglingName: coverage of thereapNamebranchserver/aws/kafka/parity_sdk_test.go(real aws-sdk-go-v2 kafka and cloudwatch clients):TestSDKListNodesClientSubnetTestSDKAsyncClusterLifecycleTestSDKAsyncVpcConnectionCreatingTestSDKKafkaMetricsAndDefaultKey: ListMetrics is empty after create. After two scheduler ticks one minute apart, GetMetricStatistics returns 2 datapoints with SampleCount 1, Maximum 1 and Unit Count. The v1 and v2 key ARNs match.providers/aws/tickables_test.gois updated for the new Kafka tickable.Commands, all passing:
Real-user test (
cloudemu serveplus AWS CLI 1.44.87 / botocore 1.42.97)With
serve -providers aws -aws-port 4607 -async-settle:With plain
serve -providers aws -aws-port 4607(default mode):ACTIVE.AVAILABLE.DELETING, and the next describe returns NotFoundException.The server was stopped after each run, and nothing was left listening on 4607.
Terraform was not run.
Known limits / risks
aws/kafkakey ARN is a deterministic UUID for each account and region. It is not created in, or resolvable through, the cloudemu KMS mock.settle.DefaultClusterSettle), not the minutes real MSK takes. Removal after DELETING happens on the next serve tick (every second by default) or the next read or create. In library mode, which never ticks, only the read or create removes it.--async-settle, rejecting updates and topic, SCRAM and policy writes on non-ACTIVE clusters is new behavior, but only under that flag.Tickreturningtrueonce a minute for each active cluster marks state as changed for the persist flusher. EFS does the same.Related Issues
Part of #1517