Create a custom rule with Function Compute

Background information

To learn about the concepts, use cases, and working principles of custom function rules, see Definition and working principles of custom function rules.

Procedure

This example writes a function that checks ECS instance CPU core counts. Instances with two or fewer CPU cores are 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 China (Shanghai) or .

   4. On the Services page, click Create Service.

   5. In the Create Service panel, enter a service Name and use defaults for other parameters.

   6. Click OK.

2. Create a function.

   1. In the service that you created in Step 1, click Create Function.

   2. On the Create Function page, enter a Function Name. Set Request Handler Type to Event Handler and Runtime to Python 3.10. Use defaults for other parameters.

   3. Click Create.

3. Configure the function code.

   1. Copy the following code and paste it into 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()
      # Compliance types.
      COMPLIANCE_TYPE_COMPLIANT = 'COMPLIANT'
      COMPLIANCE_TYPE_NON_COMPLIANT = 'NON_COMPLIANT'
      COMPLIANCE_TYPE_NOT_APPLICABLE = 'NOT_APPLICABLE'
      # Resource configuration delivery types.
      CONFIGURATION_TYPE_COMMON = 'COMMON'
      CONFIGURATION_TYPE_OVERSIZE = 'OVERSIZE'
      CONFIGURATION_TYPE_NONE = 'NONE'
      
      
      # The entry point function that orchestrates and processes business logic.
      def handler(event, context):
          """
          The handler function.
          :param event: The event data.
          :param context: The context object.
          :return: The evaluation results.
          """
          # The input parameters of the function.
          logger.info(f'Printing function input parameters:{event}')
      
          # Validate the event. You can copy this code block directly.
          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')
      
          # The resource configuration. This parameter has a value when the trigger is set to Configuration Change.
          # When you create or manually run a rule, Cloud Config invokes the function to evaluate all specified resources one by one.
          # If the configuration of a resource changes, Cloud Config automatically invokes the function to evaluate the resource based on the change.
          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 delivered resource configuration exceeds the upper limit (100 KB).
          # If it does, you must call the API for querying resource details to obtain 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 the resource. You must implement the evaluation logic based on your business requirements. The following code is for reference only.
          compliance_type, annotation = evaluate_configuration_item(
              rule_parameters, configuration_item)
      
          # Set the evaluation results. The format must comply with the following example.
          evaluations = [
              {
                  'accountId': account_id,
                  'complianceResourceId': resource_id,
                  'complianceResourceName': resource_name,
                  'complianceResourceType': resource_type,
                  'complianceRegionId': region_id,
                  'orderingTimestamp': ordering_timestamp,
                  'complianceType': compliance_type,
                  'annotation': annotation
              }
          ]
      
          # Submit the evaluation results to Cloud Config. You can copy this code block directly.
          put_evaluations(creds, result_token, evaluations)
          return evaluations
      
      
      # Evaluate the number of CPU cores in the ECS instance.
      def evaluate_configuration_item(rule_parameters, configuration_item):
          """
          The evaluation logic.
          :param rule_parameters: The parameters of the rule.
          :param configuration_item: The configuration item.
          :return: The compliance evaluation result.
          """
          # Initialize the return values.
          compliance_type = COMPLIANCE_TYPE_COMPLIANT
          annotation = None
      
          # Obtain the complete configuration of the resource.
          full_configuration = configuration_item['configuration']
          if not full_configuration:
              annotation = 'Configuration is empty.'
              return compliance_type, annotation
      
          # Convert the configuration to the JSON format.
          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):
          """
          Validate the event.
          :param event: The event data.
          :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):
          """
          Parse a JSON string.
          :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
      
      
      # Submit the evaluation results to Cloud Config. You can copy this code block directly.
      def put_evaluations(creds, result_token, evaluations):
          """
          Submits the evaluation results to Cloud Config.
          :param context: The context of Function Compute.
          :param result_token: The callback token.
          :param evaluations: The evaluation results.
          :return: None
          """
          # The Function Compute service role must have the AliyunConfigFullAccess permission.
          client = AcsClient(creds.access_key_id, creds.access_key_secret, region_id='cn-shanghai')
      
          # Create a request and set its parameters. The domain is config.cn-shanghai.aliyuncs.com.
          request = CommonRequest()
          request.set_domain('config.cn-shanghai.aliyuncs.com')
          request.set_version('2020-09-07')
          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

      This code checks CPU core counts. To find parameter names for your rule, check the resource configuration (configuration) to obtain the parameter name for your rule. View resource information, Step 6. For example, in the Configuration Details section of an ECS instance, click View JSON. The CPU core count parameter is Cpu. Obtain the expected CpuCount value from rule_parameters.

      The function submits evaluation results through the PutEvaluations API. Note the DeleteMode field. If enabled, evaluation results not updated during the current evaluation are automatically deleted.

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

4. Create a custom rule by using 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. On the Select Create Method page, select Based on Function Compute, select Function ARN, and then click Next.

      For the Function ARN, set Region to China (Shanghai) or . For Service, select the service that you created in Step 1. For Function, select the function that you created in Step 2.

   6. On the Set Basic Properties page, enter a Rule Name and click Add Rule Parameter. Enter CpuCount for Rule Parameters and 2 for Expected Value. Then, set Trigger to Configuration Changes and click Next.

      Note

      • The rule parameter name must match the rule_parameters key in the code in Step 3.

      • When you create, modify, or re-evaluate a rule, Cloud Config sends the configurations of all resources of the specified type to Function Compute for evaluation.

   7. On the Set Effective Scope page, set Resource Type to ECS Instance and click Next.

      Note

      • To avoid invalid evaluations, select only the resource types that you want to evaluate.

      • The rule evaluates all resources of the associated type in your account.

   8. On the Set Correction page, click Submit.

      Note

      You can turn on the Set Correction switch and configure a custom remediation. Set a custom remediation.

5. View detection results.

   The Rules list shows the number of non-compliant resources detected by the rule.

   Note

   Click a Rule Identifier or click Details in the Actions column to view the Compliance Result of Related Resources.

   If the Non-compliant Resources column for the rule displays No Data, but matching resources exist in your account, the function invocation or evaluation submission may have failed. On the Invocation Log tab of the function, click Request Log in the Actions column to troubleshoot. View invocation logs.

Related topics

• Configure a custom rule

• Data structure of a custom rule

• Custom rule function not triggered

• Evaluation result is No Data

• Rule not evaluated