How to configure EnhancedMonitoring for MSK express broker through cdk/cloudformation

Hello,

I see that you are facing the error below while updating the EnhancedMonitoring from "DEFAULT" to "PER_BROKER" in your MSK Cluster via CDK/CloudFormation.

Broker logs are not supported for clusters with Express instance types.

Please note that when updating enhancedMonitoring from DEFAULT to PER_BROKER on Express clusters, CloudFormation's UpdateMonitoring API validation automatically includes logging parameters in the request, which Express brokers reject. Therefore it is not possible via CDK/CloudFormation yet.

"loggingInfo": { "brokerLogs": { "cloudWatchLogs": { "enabled": false }, "firehose": { "enabled": false }, "s3": { "enabled": false } } },

However if you wish to achieve this via CDK then you would have to make use of lambda backed custom resources [1] wherein the UpdateMonitoring API call can be made via lambda function to work around it during cdk deploy.

//MSK Cluster
 const expressCluster = new msk.CfnCluster(this, 'ExpressCluster', {
      clusterName: 'test-express-cluster',
      kafkaVersion: '3.8.x',
      numberOfBrokerNodes: 6,
      brokerNodeGroupInfo: {
        instanceType: 'express.m7g.large', 
        clientSubnets: [...],
      },
      enhancedMonitoring: 'DEFAULT', 
    });
    
    // Lambda function for custom resource
    const mskUpdaterFunction = new lambda.Function(this, 'MskUpdaterFunction', {
      runtime: lambda.Runtime.NODEJS_20_X,
      handler: 'index.handler',
      code: lambda.Code.fromInline(`
        const { KafkaClient, UpdateMonitoringCommand, DescribeClusterCommand } = require('@aws-sdk/client-kafka');
        const https = require('https');
        const url = require('url');
    
        const kafka = new KafkaClient({});
    
        exports.handler = async (event) => {
          console.log('Event:', JSON.stringify(event, null, 2));
          
          try {
            const { RequestType, ResourceProperties } = event;
            const { ClusterArn, EnhancedMonitoring } = ResourceProperties;
            
            if (RequestType === 'Delete') {
              await sendResponse(event, 'SUCCESS', { PhysicalResourceId: ClusterArn });
              return;
            }
            
            // Get current version
            const describeResponse = await kafka.send(new DescribeClusterCommand({ ClusterArn }));
            const currentVersion = describeResponse.ClusterInfo.CurrentVersion;
            
            // Update monitoring
            const updateResponse = await kafka.send(new UpdateMonitoringCommand({
              ClusterArn,
              CurrentVersion: currentVersion,
              EnhancedMonitoring
            }));
            
            await sendResponse(event, 'SUCCESS', {
              PhysicalResourceId: ClusterArn,
              Data: { ClusterOperationArn: updateResponse.ClusterOperationArn }
            });
            
          } catch (error) {
            console.error('Error:', error);
            await sendResponse(event, 'FAILED', {
              PhysicalResourceId: event.ResourceProperties.ClusterArn || 'unknown',
              Reason: error.message
            });
          }
        };
    
        async function sendResponse(event, status, data) {
          const responseBody = JSON.stringify({
            Status: status,
            Reason: data.Reason || 'Success',
            PhysicalResourceId: data.PhysicalResourceId,
            StackId: event.StackId,
            RequestId: event.RequestId,
            LogicalResourceId: event.LogicalResourceId,
            Data: data.Data || {}
          });
    
          const parsedUrl = url.parse(event.ResponseURL);
          const options = {
            hostname: parsedUrl.hostname,
            port: 443,
            path: parsedUrl.path,
            method: 'PUT',
            headers: {
              'Content-Type': '',
              'Content-Length': responseBody.length
            }
          };
    
          return new Promise((resolve, reject) => {
            const req = https.request(options, (res) => {
              console.log('Response status:', res.statusCode);
              resolve();
            });
            req.on('error', reject);
            req.write(responseBody);
            req.end();
          });
        }
      `),
      timeout: cdk.Duration.minutes(5)
    });
    
    // Grant permissions
    mskUpdaterFunction.addToRolePolicy(new iam.PolicyStatement({
      actions: ['kafka:UpdateMonitoring', 'kafka:DescribeCluster'],
      resources: ['*']
    }));
    
    // Custom resource
    const mskUpdater = new cdk.CustomResource(this, 'MskMonitoringUpdater', {
      serviceToken: mskUpdaterFunction.functionArn,
      properties: {
        ClusterArn: expressCluster.attrArn,
        EnhancedMonitoring: 'PER_BROKER'
      }
    });
    
    mskUpdater.node.addDependency(expressCluster);

Kindly note that this might not be a viable approach however this is the only way to achieve your use case via CDK. I would recommend creating a new MSK cluster with enhancedMonitoring: 'PER_BROKER' instead.

REFERENCES:

[1] https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.CustomResource.html

For Amazon MSK Express brokers, you can configure enhanced monitoring through CloudFormation or CDK by setting the enhancedMonitoring property to one of the supported monitoring levels: DEFAULT, PER_BROKER, PER_TOPIC_PER_BROKER, or PER_TOPIC_PER_PARTITION.

The error you're encountering about broker logs not being supported for Express instance types suggests that even though you're not explicitly configuring broker logs, the CloudFormation might be attempting to apply a default logging configuration that's incompatible with Express brokers.

To resolve this issue, try explicitly setting the loggingInfo property to undefined or omit it entirely from your CDK code:

const mskCluster = new msk.CfnCluster(this, 'ProvisionedCluster', {
clusterName: ...,
kafkaVersion: ...,
numberOfBrokerNodes: ...,
brokerNodeGroupInfo: {
instanceType: ...,
clientSubnets: ...,
securityGroups: [mskSG.securityGroupId],
connectivityInfo: {
...
},
},
encryptionInfo: {
...
},
clientAuthentication: {
...
},
enhancedMonitoring: "PER_BROKER",
// Do not include loggingInfo property at all
});

If you're still encountering issues, you might need to check if your instance type is correctly specified as an Express broker type, as different monitoring configurations apply to Provisioned clusters versus Express clusters.

Remember that for MSK clusters, the enhanced monitoring levels determine which metrics are collected:

• DEFAULT provides basic metrics
• PER_BROKER adds broker-specific metrics
• PER_TOPIC_PER_BROKER adds topic-level metrics for each broker
• PER_TOPIC_PER_PARTITION provides the most detailed metrics

Note that according to AWS Config rules, for compliance with the msk-enhanced-monitoring-enabled rule, the monitoring level should be set to PER_TOPIC_PER_BROKER or PER_TOPIC_PER_PARTITION.
Sources
msk-enhanced-monitoring-enabled - AWS Config
Amazon MSK metrics for monitoring Express brokers with CloudWatch - Amazon Managed Streaming for Apache Kafka
Create a provisioned Amazon MSK cluster using the AWS CLI - Amazon Managed Streaming for Apache Kafka
Amazon Managed Streaming for Apache Kafka (Amazon MSK) controls - AWS Control Tower