DevOps process enrichment development guide

Developing entity and relational data follows a standard flow:

Example of DevOps scenario development

This section uses a common microservice DevOps scenario to demonstrate the complete development flow for entities and relationships.

├── devops_data_generator/           # Main directory for the data generator
│   ├── config/                      # Configuration files
│   │   ├── app_config.yaml          # Application configuration (API credentials, SLS configuration, etc.)
│   │   ├── data_mapping.yaml        # Entity/relationship field mapping configuration
│   │   ├── static_topo.yaml         # Static relationship configuration
│   │   └── repo_image_mapping.yaml  # Code repository to image repository mapping
│   ├── tasks/                       # Task implementations
│   │   ├── developer_task.py        # Developer entity generation
│   │   ├── code_repository_task.py  # Code repository entity generation
│   │   ├── code_release_task.py     # Code release entity generation
│   │   ├── image_registry_task.py   # Image repository entity generation
│   │   ├── image_task.py            # Image entity generation
│   │   ├── kubernetes_pod_task.py   # Pod entity generation
│   │   ├── developer_manages_code_repository_task.py  # "manages" relationship generation
│   │   ├── code_release_sourced_from_code_repository_task.py  # "sourced_from" relationship for releases
│   │   ├── image_sourced_from_code_release_task.py    # "sourced_from" relationship for images
│   │   ├── image_registry_contains_image_task.py      # "contains" relationship
│   │   ├── pod_uses_image_task.py   # "uses" relationship generation
│   │   └── static_topo_task.py      # Static relationship processing
│   ├── generator/                   # Data generator
│   │   └── sls_data_generator.py    # SLS data format transformation
│   ├── sender/                      # Data sender
│   │   └── sls_data_sender.py       # SLS data upload
│   ├── shared/                      # Shared components
│   │   └── data_context.py          # Data sharing between tasks
│   ├── orchestrator.py              # Task orchestrator
│   ├── main.py                      # Command line entry point
│   ├── app.py                       # Flask API entry point
│   ├── Dockerfile                   # Docker build file
│   ├── Makefile                     # Quick operation commands
│   └── requirements.txt             # Python dependencies
└── umodel/                          # UModel schema definitions
    ├── entity_set/                  # Entity definitions
    │   ├── devops_devops.developer.yaml
    │   ├── devops_devops.code_repository.yaml
    │   ├── devops_devops.code_release.yaml
    │   ├── devops_devops.image_registry.yaml
    │   └── devops_devops.image.yaml
    └── entity_set_link/             # Relationship definitions
        ├── devops.developer_manages_devops.code_repository.yaml
        ├── devops.code_release_sourced_from_devops.code_repository.yaml
        ├── devops.image_sourced_from_devops.code_release.yaml
        ├── devops.image_registry_contains_devops.image.yaml
        └── k8s.pod_uses_devops.image.yaml

Step 1: Define the schema

1. Define the developer entity schema.

   Developer entity schema

   Important

   The id_generator is the most critical configuration in UModel modeling. It determines the unique identity of the entity.

   metadata:
       description:
           en_us: Developer
           zh_cn: Developer
       display_name:
           en_us: Developer
           zh_cn: Developer
       domain: devops
       kind: entity_set
       name: devops.developer
   
   spec:
       id_generator: lower(to_hex(md5(cast(work_no as varbinary))))
       fields:
           - description:
               en_us: Work Number
               zh_cn: Work Number
             display_name:
               en_us: Work Number
               zh_cn: Work Number
             filterable: true
             name: work_no
             orderable: true
             short_description:
               en_us: Work Number
               zh_cn: Work Number
             type: string
           - description:
               en_us: Developer Name
               zh_cn: Developer Name
             display_name:
               en_us: Developer Name
               zh_cn: Developer Name
             filterable: true
             name: name
             orderable: true
             short_description:
               en_us: Developer Name
               zh_cn: Developer Name
             type: string
             ....
       name_fields:
           - name
           - work_no
           - email
           - team
           - role
       primary_key_fields:
           - work_no
       time_field: __time__
       type: entity_set

2. Define the manages relationship schema.

   Manages relationship schema

   # EntitySetLink: manages
   metadata:
       description:
           en_us: |
               The link between "devops.developer" and "devops.code_repository".
           zh_cn: Developer-Manages-Code-Repository
       display_name:
           en_us: |
               Developer-Manages-Code-Repository
           zh_cn: Developer-Manages-Code-Repository
       domain: devops
       kind: entity_set_link
       name: devops.developer_manages_devops.code_repository
   # Source and destination of the relationship
   spec:
       dest:
           domain: devops
           kind: entity_set
           name: devops.code_repository
       entity_link_type: manages
       priority: 5
       src:
           domain: devops
           kind: entity_set
           name: devops.developer

Step 2: Implement data collection

1. Retrieve raw data for the developer entity from the data source, determine the upload policy, and write the code.

   • Data source and collection method: Collect data from the DevOps API.

     • Data source: Alibaba Cloud DevOps API - ListRepositoryMemberWithInherited.

     • Collection method: Traverse all code repositories, retrieve the members, and remove duplicates.

   • Upload policy: Because personnel changes are infrequent, the data volume is small (only a few hundred people), and the cost of a full upload is low, the recommended policy is a scheduled full upload.

     • Execution frequency: Because the data changes slowly, once a day is sufficient.

     • KeepAlive: 2 days (172,800 seconds). This is twice the full upload epoch to prevent data loss if a task fails.

     Example code for developer entity data collection

     Note

     Note the following:

     • Use a dictionary to deduplicate data and avoid repeating the same developer.

     • Record logs for easy debugging and monitoring.

     • Handle API paging if needed.

     • Set KeepAlive to 2 days to ensure data remains queryable if the scheduled task fails.

     from alibabacloud_devops20210625.client import Client
     from alibabacloud_devops20210625 import models
     import schedule
     
     def fetch_developers():
         """Fetch all developers from the DevOps API"""
         # Initialize the DevOps client
         client = Client(
             access_key_id=config['devops']['access_key_id'],
             access_key_secret=config['devops']['access_key_secret']
         )
         
         developers = {}  # Use a dictionary to remove duplicates
         
         # Get all code repositories
         list_repo_request = models.ListRepositoriesRequest(
             organization_id=config['devops']['organization_id']
         )
         repositories = client.list_repositories(list_repo_request)
         
         # Traverse each repository to get its members
         for repo in repositories.body.result:
             list_member_request = models.ListRepositoryMemberWithInheritedRequest(
                 organization_id=config['devops']['organization_id'],
                 repository_id=repo.id
             )
             
             members = client.list_repository_member_with_inherited(
                 list_member_request
             )
             
             # Collect member information (and remove duplicates)
             for member in members.body.result:
                 user_id = member.aliyun_pk  # Use the Aliyun PK as the unique identifier
                 if user_id not in developers:
                     developers[user_id] = {
                         'user_id': user_id,
                         'name': member.name,
                         'email': member.email,
                         'role': member.role_name
                     }
         
         logger.info(f"Collected {len(developers)} developers")
         return list(developers.values())
     
     def sync_developers():
         """Perform a scheduled full synchronization of developers"""
         logger.info("Starting to sync developers...")
         
         # 1. Collect data
         raw_developers = fetch_developers()
         
         # 2. Transform data (explained in Step 3)
         entities = convert_all_developers(raw_developers)
         
         # 3. Upload data (explained in Step 4)
         success = upload_entities(entities, workspace='o11y-workspace')
         
         if success:
             logger.info(f"Successfully uploaded {len(entities)} developer entities")
         else:
             logger.error("Upload failed")
     
     # Set a scheduled task: run at 3:00 AM every day
     schedule.every().day.at("03:00").do(sync_developers)

2. Retrieve raw data for the manages relationship from the data source, determine the upload policy, and write the code.

   • Data source and collection method: Retrieve the relationship where developers manage code repositories.

     • Data source: A YAML configuration file that defines the relationship (developer to code repository).

     • Collection method: Associate static configurations using existing entity IDs.

   • Upload policy: Because the manages relationship is relatively stable, changes infrequently, and the data source is a manually maintained static configuration file, the recommended policy is a scheduled full upload.

     • Execution frequency: Once per hour to promptly sync configuration changes.

     • KeepAlive: 2 hours. This value is twice the full cycle period to prevent task failures.

     Example code for manages relationship data collection

     Example configuration file:

     # Developer to code repository management mapping
     manage_mappings:
       "1711055":  # Work number (primary key of the developer)
         repositories:
           - "mymall-order"      # Repository name (identifier of the repository)
           - "mymall-cart"
       
       "1703317":
         repositories:
           - "mymall-product"
           - "mymall-member"
           - "mymall-payment"

     Implementation code:

     import yaml
     import schedule
     
     def generate_manage_relationships():
         """Generate developer-manages-code-repository relationships based on the configuration file"""
         # 1. Load the configuration file
         with open('config/manage_mapping.yaml', 'r') as f:
             config = yaml.safe_load(f)
         
         # 2. Query existing entities (prerequisite: entities are already uploaded to EntityStore)
         developers = query_entities_from_sls('devops.developer')
         repositories = query_entities_from_sls('devops.code_repository')
         
         # 3. Build a lookup index (for fast lookups by primary key)
         dev_map = {d['work_no']: d for d in developers}
         repo_map = {r['repo_name']: r for r in repositories}
         
         # 4. Generate relationship data
     
         relationships = [ ]
     
         for work_no, mapping in config['manage_mappings'].items():
             # Check if the developer exists
             if work_no not in dev_map:
                 logger.warning(f"Developer {work_no} does not exist, skipping")
                 continue
             
             developer = dev_map[work_no]
             
             # Traverse all repositories managed by this developer
             for repo_name in mapping['repositories']:
                 # Check if the code repository exists
                 if repo_name not in repo_map:
                     logger.warning(f"Code repository {repo_name} does not exist, skipping")
                     continue
                 
                 repository = repo_map[repo_name]
                 
                 # Build relationship data (keep original identifiers for use in the transformation step)
                 relationships.append({
                     'src_entity_id': developer['__entity_id__'],
                     'dest_entity_id': repository['__entity_id__'],
                     'work_no': work_no,
                     'repo_name': repo_name
                 })
         
         logger.info(f"Generated {len(relationships)} manages relationships")
         return relationships
     
     def sync_manage_relationships():
         """Perform a scheduled full synchronization of manages relationships"""
         logger.info("Starting to sync manages relationships...")
         
         # 1. Generate relationship data
         raw_relationships = generate_manage_relationships()
         
         # 2. Transform data (explained in Step 3)
         relationships = [convert_to_relationship(r) for r in raw_relationships]
         
         # 3. Upload data (explained in Step 4)
         success = upload_relationships(relationships, workspace='o11y-workspace')
         
         if success:
             logger.info(f"Successfully uploaded {len(relationships)} manages relationships")
         else:
             logger.error("Upload failed")
     
     # Set a scheduled task: run every hour
     schedule.every().hour.do(sync_manage_relationships)

Step 3: Transform the data

1. Write code to transform the raw data into the EntityStore format.

   Developer entity: Data transformation

   import hashlib
   import time
   
   def convert_developer_to_entity(raw_data):
       """Transform raw developer data into the entity format"""
       # 1. Generate the entity_id (based on the id_generator rule in the schema)
       work_no = raw_data['user_id']
       entity_id = hashlib.md5(work_no.encode()).hexdigest()
       
       # 2. Build the entity data
       entity = {
           # Required system fields
           '__domain__': 'devops',
           '__entity_type__': 'devops.developer',
           '__entity_id__': entity_id,
           '__method__': 'Update',  # Use Update for full synchronization
           '__last_observed_time__': str(int(time.time())),
           '__keep_alive_seconds__': str(86400 * 2),  # 2 days
           
           # Business fields
           'work_no': work_no,
           'name': raw_data['name'],
           'email': raw_data['email'],
           'role': raw_data.get('role', 'developer')
       }
       
       return entity
   
   # Batch transformation
   def convert_all_developers(raw_developers):
       """Transform developer data in batches"""
   
       entities = [ ]
   
       for raw_data in raw_developers:
           try:
               entity = convert_developer_to_entity(raw_data)
               entities.append(entity)
           except Exception as e:
               logger.error(f"Transformation failed: {raw_data}, Error: {e}")
       
       return entities

   Manages relationship: Data transformation

   def convert_to_relationship(rel_data):
       """Transform relationship data into the standard format"""
       return {
           # Relationship system fields
           '__src_domain__': 'devops',
           '__src_entity_type__': 'devops.developer',
           '__src_entity_id__': rel_data['src_entity_id'],
           '__dest_domain__': 'devops',
           '__dest_entity_type__': 'devops.code_repository',
           '__dest_entity_id__': rel_data['dest_entity_id'],
           '__link_type__': 'manages',
           '__method__': 'Update',
           '__last_observed_time__': str(int(time.time())),
           '__keep_alive_seconds__': str(7200),  # 2 hours
           
           # Relationship business fields (optional)
           'work_no': rel_data.get('work_no'),
           'repo_name': rel_data.get('repo_name')
       }

Step 4: Upload the data

1. Use the Simple Log Service (SLS) SDK to upload data to the destination Logstore.

   Initialize the SLS client

   You can retrieve parameters as follows:

   • To obtain an AccessKey ID and AccessKey secret, see Create an AccessKey.

   • To obtain the endpoint:

     1. Log on to the Simple Log Service console. In the Project list, click the destination project.

     2. Click next to the project name to go to the project overview page. In the Endpoints section, copy the public endpoint.

   from aliyun.log import LogClient
   
   def init_sls_client():
       """Initialize the SLS client"""
       return LogClient(
           endpoint=config['sls']['endpoint'],
           accessKeyId=config['sls']['access_key_id'],
           accessKey=config['sls']['access_key_secret']
       )

2. Upload the entity and relationship data.

   Important

   • Upload data in batches to improve performance, for example, 1,000 records per batch.

   • All field values must be converted to strings.

   • Add exception handling and log recording.

   Upload entity data

   from aliyun.log import PutLogsRequest, LogItem
   
   def upload_entities(entities, workspace):
       """Upload entity data to EntityStore in batches"""
       client = init_sls_client()
       
       project = config['sls']['project']
       logstore = f"{workspace}__entity"
       
       # Upload in batches (1,000 records per batch)
       batch_size = 1000
       for i in range(0, len(entities), batch_size):
           batch = entities[i:i+batch_size]
           
           # Build a list of LogItems
   
           log_items = [ ]
   
           for entity in batch:
               log_item = LogItem()
               log_item.set_time(int(time.time()))
               log_item.set_contents([
                   (key, str(value)) for key, value in entity.items()
               ])
               log_items.append(log_item)
           
           # Upload
           try:
               request = PutLogsRequest(
                   project=project,
                   logstore=logstore,
                   topic="",
                   logitems=log_items
               )
               response = client.put_logs(request)
               logger.info(f"Batch {i//batch_size + 1} uploaded successfully")
           except Exception as e:
               logger.error(f"Batch {i//batch_size + 1} upload failed: {e}")
               return False
       
       return True

   Upload relationship data

   def upload_relationships(relationships, workspace):
       """Upload relationship data to EntityStore in batches"""
       client = init_sls_client()
       
       project = config['sls']['project']
       logstore = f"{workspace}__topo"  # Upload relationships to __topo
       
       # Upload in batches
       batch_size = 1000
       for i in range(0, len(relationships), batch_size):
           batch = relationships[i:i+batch_size]
           
   
           log_items = [ ]
   
           for rel in batch:
               log_item = LogItem()
               log_item.set_time(int(time.time()))
               log_item.set_contents([
                   (key, str(value)) for key, value in rel.items()
               ])
               log_items.append(log_item)
           
           try:
               request = PutLogsRequest(
                   project=project,
                   logstore=logstore,
                   topic="",
                   logitems=log_items
               )
               response = client.put_logs(request)
               logger.info(f"Relationship batch {i//batch_size + 1} uploaded successfully")
           except Exception as e:
               logger.error(f"Relationship batch {i//batch_size + 1} upload failed: {e}")
               return False
       
       return True

Related reference

Data collection for other entities

The DevOps scenario includes multiple entity types. Different collection methods are used for different data sources:

DevOps entities (collected from the Alibaba Cloud DevOps API)

Container image entities (collected from Alibaba Cloud ACR)

K8s entities (collected from CloudMonitor (CMS))

Generation methods for other relationships

Relationships in the DevOps scenario can be divided into static relationships and dynamic relationships based on their generation method:

Static relationships (based on configuration files)

Dynamic relationships (based on data association)

Reference for entity and relationship upload policies