Cloud Config:Create a custom rule based on Function Compute

Background information

For more information about the concept, usage scenarios, and principles of custom rule functions, see Definition and implementation of custom rule functions.

Procedure

This section describes how to write function code to check the number of CPU cores of an ECS instance. If the number of CPU cores of the ECS instance is less than or equal to 2, the evaluation result is compliant. If the number of CPU cores of the ECS instance is greater than 2, the evaluation result is non-compliant.

1. Create a service.

   1. Log on to the Function Compute console.

   2. In the left-side navigation pane, click Services & Functions.

   3. In the top navigation bar, select a region, such as Singapore (Singapore).

   4. On the Services page, click Create Service.

   5. In the Create Service panel, enter a service name in the Name field and retain the default values for other parameters.

   6. Click OK.

2. Create a function.

   1. Click the name of the service that you created in Step 1. On the page that appears, click Create Function.

   2. On the Create Function page, configure Function Name, and set Handler Type to Event Handler and Runtime to Python 3.10. Retain the default values for other parameters.

   3. Click Create.

3. Configure the function code.

   1. Copy and paste the following code to the index.py file:

      # # !/usr/bin/env python
      # # -*- encoding: utf-8 -*-
      import json
      import logging
      
      from aliyunsdkcore.client import AcsClient
      from aliyunsdkcore.request import CommonRequest
      
      
      logger = logging.getLogger()
      # The compliance evaluation results.
      COMPLIANCE_TYPE_COMPLIANT = 'COMPLIANT'
      COMPLIANCE_TYPE_NON_COMPLIANT = 'NON_COMPLIANT'
      COMPLIANCE_TYPE_NOT_APPLICABLE = 'NOT_APPLICABLE'
      # The types of pushed configurations.
      CONFIGURATION_TYPE_COMMON = 'COMMON'
      CONFIGURATION_TYPE_OVERSIZE = 'OVERSIZE'
      CONFIGURATION_TYPE_NONE = 'NONE'
      
      
      # The entry function that orchestrates and processes business logic. 
      def handler(event, context):
          """
          Process an event.
          :param event: the event.
          :param context: the context.
          :return: the compliance evaluation result.
          """
          # The input parameter of the function.
          logger.info(f'Print the input parameter of the function:{event}')
      
          # Check whether the event is valid. You can copy the following sample code based on your business requirements. 
          evt = validate_event(event)
          if not evt:
              return None
          creds = context.credentials
          rule_parameters = evt.get('ruleParameters')
          result_token = evt.get('resultToken')
          invoking_event = evt.get('invokingEvent')
          ordering_timestamp = evt.get('orderingTimestamp')
      
          # Obtain the resource configurations. If a rule is set to be triggered upon configuration changes, the input parameters are specified. After a rule that you create is triggered or you manually trigger a rule, Cloud Config invokes the rule function to evaluate associated resources one by one. If the configurations of a resource change, Cloud Config automatically invokes the related rule function to evaluate the resource. 
          configuration_item = invoking_event.get('configurationItem')
          account_id = configuration_item.get('accountId')
          resource_id = configuration_item.get('resourceId')
          resource_type = configuration_item.get('resourceType')
          region_id = configuration_item.get('regionId')
          resource_name = configuration_item.get('resourceName')
      
          # Check whether the size of the pushed resource configurations is greater than 100 KB. If yes, call an API operation to query the complete data. 
          configuration_type = invoking_event.get('configurationType')
          if configuration_type and configuration_type == CONFIGURATION_TYPE_OVERSIZE:
              resource_result = get_discovered_resource(creds, resource_id, resource_type, region_id)
              resource_json = json.loads(resource_result)
              configuration_item["configuration"] = resource_json["DiscoveredResourceDetail"]["Configuration"]
      
          # Evaluate resources. You must implement the evaluation logic based on your business scenario. The following code is only for reference: 
          compliance_type, annotation = evaluate_configuration_item(
              rule_parameters, configuration_item)
      
          # Specify the compliance evaluation result. The result must use the syntax shown in the following sample: 
          evaluations = [
              {
                  'accountId': account_id,
                  'complianceResourceId': resource_id,
                  'complianceResourceName': resource_name,
                  'complianceResourceType': resource_type,
                  'complianceRegionId': region_id,
                  'orderingTimestamp': ordering_timestamp,
                  'complianceType': compliance_type,
                  'annotation': annotation
              }
          ]
      
          # Return the compliance evaluation result and write the result to Cloud Config. You can directly use the following sample code in your function code. 
          put_evaluations(creds, result_token, evaluations)
          return evaluations
      
      
      # Evaluate the number of CPU cores of the ECS instance. 
      def evaluate_configuration_item(rule_parameters, configuration_item):
          """
          Evaluate a resource.
          :param rule_parameters: the input parameters of the rule.
          :param configuration_item: the configurations of the resource.
          :return: the compliance evaluation result.
          """
          # Initialize the return values.
          compliance_type = COMPLIANCE_TYPE_COMPLIANT
          annotation = None
      
          # Obtain the complete resource configurations.
          full_configuration = configuration_item['configuration']
          if not full_configuration:
              annotation = 'Configuration is empty.'
              return compliance_type, annotation
      
          # Convert the configurations to a JSON object.
          configuration = parse_json(full_configuration)
          cpu_count = configuration.get('Cpu')
          eq_count = rule_parameters.get('CpuCount')
          if cpu_count and cpu_count <= int(eq_count):
            annotation = json.dumps({"configuration":cpu_count,"desiredValue":eq_count,"operator":"Greater","property":"$.Cpu"})
            compliance_type = COMPLIANCE_TYPE_NON_COMPLIANT
            return compliance_type, annotation
          return compliance_type, annotation
      
      
      def validate_event(event):
          """
          Check whether an event is valid.
          :param event: the event.
          :return: the JSON object.
          """
          if not event:
              logger.error('Event is empty.')
          evt = parse_json(event)
          logger.info('Loading event: %s .' % json.dumps(evt))
      
          if 'resultToken' not in evt:
              logger.error('ResultToken is empty.')
              return None
          if 'ruleParameters' not in evt:
              logger.error('RuleParameters is empty.')
              return None
          if 'invokingEvent' not in evt:
              logger.error('InvokingEvent is empty.')
              return None
          return evt
      
      
      def parse_json(content):
          """
          Convert a JSON string into a JSON object.
          :param content: the JSON string.
          :return: the JSON object.
          """
          try:
              return json.loads(content)
          except Exception as e:
              logger.error('Parse content:{} to json error:{}.'.format(content, e))
              return None
      
      
      # Return the compliance evaluation result and write the result to Cloud Config. You can directly use the following sample code in your function code. 
      def put_evaluations(creds, result_token, evaluations):
          """
          Call the Cloud Config API to return the compliance evaluation result and write the result to Cloud Config.
          :param context: the Function Compute context.
          :param result_token: the result token.
          :param evaluations: the compliance evaluation result.
          :return: None
          """
          # Your account must be assigned a service role of Function Compute to which the AliyunConfigFullAccess policy is attached. 
          client = AcsClient(creds.access_key_id, creds.access_key_secret, region_id='cn-shanghai')
      
          # Create a request and configure the parameters. Set the Domain parameter to config.cn-shanghai.aliyuncs.com. 
          request = CommonRequest()
          request.set_domain('config.cn-shanghai.aliyuncs.com')
          request.set_version('2019-01-08')
          request.set_action_name('PutEvaluations')
          request.add_body_params('ResultToken', result_token)
          request.add_body_params('Evaluations', evaluations)
          request.add_body_params('SecurityToken', creds.security_token)
          request.set_method('POST')
      
          try:
              response = client.do_action_with_exception(request)
              logger.info('PutEvaluations with request: {}, response: {}.'.format(request, response))
          except Exception as e:
              logger.error('PutEvaluations error: %s' % e)

      Note

      The preceding code is used to check the number of CPU cores. To obtain the parameter names of the rules that you want to configure, we recommend that you view the core parameters of the resource configuration that is specified by the configuration parameter in the code in the Cloud Config console. For more information, see Step 6 in View the information about a resource. For example, you can click the name of the desired ECS instance in the Resource ID / Name column on the Global Resources page. In the Core configuration information section of the Resource Info tab of the ECS instance, click View JSON to view the Cpu parameter that specifies the number of CPU cores. You can obtain the expected value of CpuCount from the rule_parameters parameter in the preceding code.

   2. In the upper-left corner of the Code tab, click Deploy.

4. Create custom rules based on Function Compute.

   1. Log on to the Cloud Config console.

   2. Optional. In the upper-left corner, select an account group.

      This operation is required only if you are using a management account of a resource directory. Otherwise, you do not need to perform the operation.

   3. In the left-side navigation pane, choose Compliance & Audit > Rules.

   4. On the Rules page, click Create Rule.

   5. In the Select Create Method step, select Based on Function Compute, specify a function in the Function ARN section, and then click Next.

      In the Function ARN section, select Singapore for the Region parameter, select the service that is created in Step 1 for the Service parameter, and then select the function that is created in Step 2 for the Function parameter.

   6. In the Set Basic Properties step, configure Rule Name, click Add Rule Parameter in the Parameter Settings section, and set Key to CpuCount and Expect Value to 2. Then, set Trigger to Configuration Change and click Next.

      Note

      • The value of the Key parameter must be the same as the value of the rule_parameters parameter in the code in Step 3.

      • When you create a rule, modify a rule, and re-audit a resource based on a rule, Cloud Config pushes all resource configurations of the desired resource type to Function Compute one by one and triggers a function to evaluate the resources.

   7. In the Set Effective Scope step, select Elastic Compute Service in the All Resource Type list and click Next.

      Note

      • We recommend that you select only the types of resources that you want to evaluate. If you select all resource types, invalid evaluation results are generated.

      • After you specify the resource types, Cloud Config evaluates all resources of the specified types based on the rule.

   8. In the Set Remediation step, click Submit.

      Note

      You can turn on Set Remediation and configure custom remediation for the rule as prompted. For more information, see Configure custom remediation.

5. View the check result of the rule for the number of CPU cores of the ECS instance.

   On the Rules page, you can view the number of non-compliant resources that are detected by the rule.

   Note

   You can click the rule ID in the Rule Identifier / Rule Name column or click Detail in the Actions column to view data in the Compliance Result of Related Resources section.

   If No Data is displayed in the NonCompliant Resource column for the desired rule on the Rules page, but you confirm that the resources specified in the rule are available in the current account, the function fails to be invoked or Function Compute fails to submit the evaluation result to Cloud Config. On the Logs tab of the function that you want to view, click Request Logs in the Actions column to find the cause of the failure. For more information, see View function invocation logs.

References

• How do I configure a custom rule?

• What is the data structure of a custom rule?

• Why was a function in a custom rule not triggered?

• Why are no evaluation results returned after I execute a custom rule?

• Why are my resources not evaluated after I configure a rule?