Event Data Sharding Practice: Implementation Based on Dataway Sink¶

This document details how to achieve intelligent sharding of event data (keyevent) through DataFlux Func HTTP Header Injection and Dataway Sinker Rule Configuration. With this solution, you can route event data with different business attributes and environmental characteristics to specified workspaces.

Solution Principle¶

Data Sharding Flow¶

Core Mechanism Explanation¶

DataFlux Func Side Injection: When event data is reported, dynamically generate the X-Global-Tags Header through Func configuration, containing key-value pairs required for sharding (e.g., env=prod).
Dataway Routing Matching: Dataway forwards events carrying specific identifiers to the corresponding workspace based on the rules defined in sinker.json.

1. Dataway Configuration¶

Before using this feature, ensure that Dataway is deployed and the Sinker sharding feature is enabled.

For Sinker configuration, refer to: Dataway Sinker Configuration Guide;

Note: The Dataway used by the built-in DataFlux Func in the deployment version is located under the utils namespace as internal-dataway.

2. DataFlux Func Configuration¶

Header Injection X-Global-Tags¶

Core Parameter Description¶

Parameter Name	Type	Description
`CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS`	list/string	Defines the sharding identifier generation rules for event data

Simple Example¶

Unify all workspace events into the "Event Central Management" workspace:

Access the Launcher console;
Go to the top right corner > Modify Application Configuration;
Find the func2Config configuration item under the func2 namespace;

Add the configuration:

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent     # Data category
    fields: df_source      # Field used for sharding, fill in the fixed identifier field of the event here

Configure Dataway Sinker rules: Modify the sinker.json configuration file to set data routing rules:

{
    "strict": true,
    "rules": [
        {
            "rules": [
                "{ df_source = 'monitor' }"
            ],
            "url": "Workspace Data Reporting Address"
        }
    ]
}

Special Field Description¶

Field Name	Description
`DF_WORKSPACE_UUID`	Workspace ID
`DF_WORKSPACE_NAME`	Workspace Name
`DF_MONITOR_CHECKER_ID`	Monitor ID
`DF_MONITOR_CHECKER_NAME`	Monitor Name

More Advanced Configurations¶

Configuration Method	Example	Description
Direct Extraction	`-host`	Extract the `host` field from the event data's `tags` or `fields`
Rename Field	`-src:service; dest:business_type`	Rename the `service` field to `business_type`
Value Mapping	`remap:{order:e-commerce business}`	Map the original value `order` to `e-commerce business`
Default Value	`default:unknown`	Use the default value when the field does not exist
Fixed Value	`- dest:env; fixed:prod`	Directly inject the fixed value `env=prod`

Global Tags Generation Rules¶

Field Name	Type	Default Value	Description
`[#].category`	string/[string]	`"*"`	Match data Category
`[#].fields`	string/dict [string]/[dict]	-	Extract data fields (including Tags and Fields); supports direct extraction and rule extraction
`[#].fields[#]`	string	-	Extract field name, and supports additional extraction fields (see below)
`[#].fields[#]`	dict	-	Extract field rules
`[#].fields[#].src`	string	-	Extract field name, and supports additional extraction fields (see below)
`[#].fields[#].dest`	string	Same as `src`	Field name written to Header after extraction
`[#].fields[#].default`	string	-	Default value written to Header when the specified field does not exist
`[#].fields[#].fixed`	string	-	Fixed value written to Header
`[#].fields[#].remap`	dict	`null`	Map and transform the extracted field value
`[#].fields[#].remap_default`	string	-	Default value when there is no corresponding mapped value during field value mapping If not specified, the original value is used If specified as `null`, this field is ignored
`[#].filter`	dict/string	`null`	Data matching filter Supports Tag filtering and filterString filtering

Custom Global Tags Generation Function ID¶

Function ID format: {script set ID}__{script ID}.{function name}

Function definition:

Parameter	Type	Description
`category`	string	Category, e.g., `"keyevent"`
`point`	dict	Single piece of data to be processed
`point.measurement`	string	Data `measurement`
`point.tags`	dict	Data `tags` content
`point.fields`	dict	Data `fields` content
`extra_fields`	dict	Additional extraction fields (see below)

Example:

point parameter value

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001",
    "ip"  : "1.2.3.4"
  },
  "fields": {
    "name": "Tom"
  }
}

extra_fields parameter value

{
  "DF_WORKSPACE_UUID"      : "wksp_xxxxx",
  "DF_MONITOR_CHECKER_ID"  : "rul_xxxxx",
  "DF_MONITOR_CHECKER_NAME": "Monitor XXXXX",
  "DF_WORKSPACE_NAME"      : "Workspace XXXXX"
}

Generation Effect Verification¶

Example of adding key:value in the Header {#example}

Write Event Data to the Same Workspace¶

Extract fields from events

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent
    fields:
      - host
      - name
      - DF_WORKSPACE_UUID

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001",
    "ip"  : "1.2.3.4"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: host=web-001,name=Tom,DF_WORKSPACE_UUID=wksp_xxxxx

Extract a single field from events

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent
    # Only 1 field can be abbreviated
    fields: host

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: host=web-001

Write All Data to the Same Workspace¶

Not writing category means processing all data

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields: DF_WORKSPACE_UUID

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: DF_WORKSPACE_UUID=wksp_xxxxx

Other Cases¶

Change field name when extracting fields

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - src : host
      dest: HOST

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: HOST=web-001

Map field values when extracting fields

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - src : result
      remap:
        OK     : ok
        success: ok
        failed : error
        failure: error
        timeout: error
      remap_default: unknown

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "result": "success"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: result=ok

Use default value when extracting fields

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - src    : result
      default: unknown

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: result=unknown

Write fixed value

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - dist : app
      fixed: dataPlatform

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: app=dataPlatform

Use Tag method to match data

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields: host
    filter:
      service: app-*
  - fields: client_ip
    filter:
      service: web-*

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host"     : "app-001",
    "client_ip": "1.2.3.4",
    "service"  : "app-user"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: host=app-001

Use filterString method to match data

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields: host
    filter: 'service:app-*'
  - fields: client_ip
    filter: 'service:web-*'

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "host"     : "app-001",
    "client_ip": "1.2.3.4",
    "service"  : "app-user"
  },
  "fields": {
    "name": "Tom"
  }
}

Example Written Header

X-Global-Tags: host=app-001

Custom function method to intercept event field prefix, suffix

Example Configuration

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS: my_script_set__my_script.make_global_tags

Example Function Located in Script Set my_script_set, Script my_script

def make_global_tags(category, point, extra_fields):
    # Only process event type data
    if category != 'keyevent':
        return

    global_tags_list = {}

    # Get name, region fields from data's fields or tags
    name   = point['fields'].get('name')   or point['tags'].get('name')
    region = point['fields'].get('region') or point['tags'].get('region')

    # Get name prefix
    if name:
        prefix = str(name).split('-')[0]
        global_tags_list['name_prefix'] = prefix

    # Get region suffix
    if region:
        suffix = str(region).split('-').pop()
        global_tags_list['region_suffix'] = suffix

    # Return
    return global_tags_list

Example Data

{
  "measurement": "keyevent",
  "tags": {
    "region"   : "cn-shanghai",
    "service"  : "app-user"
  },
  "fields": {
    "name": "Tom-Jerry"
  }
}

Example Written Header

X-Global-Tags: name_prefix=Tom,region_suffix=shanghai

Report Event Example:

{
  "measurement": "keyevent",
  "tags": { "host": "web-01", "service": "order" },
  "fields": { "message": "User order exception" }
}

Generated HTTP Header:

X-Global-Tags: host=web-01,business_type=e-commerce business,DF_WORKSPACE_UUID=wksp_123

3. Dataway Sinker Rule Configuration¶

Rule File Example (`sinker.json`)¶

{
  "strict": false,
  "rules": [
    {
      "rules": ["{ business_type = 'e-commerce business' }"],  // Match e-commerce business events
      "url": "https://kodo.truewatch.com?token=tkn_e-commerce space token"
    },
    {
      "rules": ["{ DF_WORKSPACE_UUID = 'wksp_123' }"],  // Match specified workspace
      "url": "https://backup.truewatch.com?token=tkn_backup space token"
    },
    {
      "rules": ["*"],  // Default rule (must exist)
      "url": "https://default.truewatch.com?token=tkn_default space token"
    }
  ]
}

Rule Syntax Explanation¶

Operator	Example	Description
`=`	`{ env = 'prod' }`	Exact match
`!=`	`{ env != 'test' }`	Not equal
`in`	`{ region in ['cn-east','cn-north'] }`	Multi-value match
`match`	`{ host match 'web-*' }`	Wildcard match

4. Datakit End Configuration Instructions¶

Basic Configuration¶

# /usr/local/datakit/conf.d/datakit.conf
[dataway]
  # Enable Sinker function
  enable_sinker = true
  # Define sharding basis fields (up to 3)
  global_customer_keys = ["host", "env"]

Precautions¶

Field Type Limitation: Only supports string type fields (all Tag values are strings)
Binary Data Support: Supports sharding of binary data such as Session Replay, Profiling
Performance Impact: Each additional sharding field increases memory usage by about 5%

5. Global Tag Impact¶

1. Global Tag Example¶

# datakit.conf
[election.tags]
    cluster = "cluster-A"  # Global election Tag
[global_tags]
    region = "cn-east"     # Global host Tag

2. Sharding Identifier Merge Logic¶

Assume the event data contains the following Tags:

{
  "tags": { "cluster": "cluster-B", "app": "payment" }
}

Final Sharding Identifier:

X-Global-Tags: cluster=cluster-B,region=cn-east

Extended Explanation: Sharding of Other Data Types¶

1. Custom Sharding Rules¶

For non-event data (e.g., logging, metric), achieve sharding by specifying category:

# Func Configuration Example: Process logging data
CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: logging
    fields:
      - src: log_level
        remap:
          error: Critical error
          warn: General warning
      - service

2. General Principles¶

Isolated Configuration: Different data categories (keyevent/logging/metric) use independent configuration blocks
Field Simplification: Sharding identifiers for a single data category do not exceed 3
Avoid Conflicts: Sharding fields of different categories are recommended to use different names

Troubleshooting¶

Common Issues¶

Phenomenon	Troubleshooting Steps
Sharding Not Effective	1. Check Dataway logs `grep 'sinker reload'` 2. Use `curl -v` to verify Header 3. Check Sinker rule priority
Partial Data Loss	1. Confirm `strict` mode status 2. Check if default rule exists
Identifier Not Injected	1. Verify Func configuration syntax 2. Check if field is string type

Diagnostic Commands¶

# View Dataway sharding statistics
curl http://localhost:9528/metrics | grep sinker_requests_total
# Manually test sharding rules
curl -X POST -H "X-Global-Tags: business_type=e-commerce business" http://dataway/v1/write/keyevent

Event Data Sharding Practice: Implementation Based on Dataway Sink¶

Solution Principle¶

Data Sharding Flow¶

Core Mechanism Explanation¶

1. Dataway Configuration¶

2. DataFlux Func Configuration¶

Header Injection X-Global-Tags¶

Core Parameter Description¶

Simple Example¶

Special Field Description¶

More Advanced Configurations¶

Global Tags Generation Rules¶

Custom Global Tags Generation Function ID¶

Generation Effect Verification¶

Write Event Data to the Same Workspace¶

Write All Data to the Same Workspace¶

Other Cases¶

3. Dataway Sinker Rule Configuration¶

Rule File Example (sinker.json)¶

Rule Syntax Explanation¶

4. Datakit End Configuration Instructions¶

Basic Configuration¶

Precautions¶

5. Global Tag Impact¶

1. Global Tag Example¶

2. Sharding Identifier Merge Logic¶

Extended Explanation: Sharding of Other Data Types¶

1. Custom Sharding Rules¶

2. General Principles¶

Troubleshooting¶

Common Issues¶

Diagnostic Commands¶

Rule File Example (`sinker.json`)¶