Event Data Sharding Practice: Implementation Based on Dataway Sink¶

This document details how to implement 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 or environmental characteristics to specified workspaces.

Solution Principle¶

Data Sharding Flow¶

Core Mechanism Explanation¶

DataFlux Func Side Identifier Injection: When event data is reported, dynamically generate the X-Global-Tags Header containing the key-value pairs required for sharding (e.g., env=prod) through Func configuration.
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 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 Plan 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 generation rules for sharding identifiers of event data

Simple Example¶

Write all workspace events uniformly to 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 configuration:

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent     # Data category
    fields: df_source      # Field used for sharding, fill in the fixed identifier field for events 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 URL"
        }
    ]
}

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 Configuration¶

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 table below)
`[#].fields[#]`	dict	-	Extract field rules
`[#].fields[#].src`	string	-	Extract field name, and supports additional extraction fields (see table below)
`[#].fields[#].dest`	string	Same as `src`	Field name written to the Header after extraction
`[#].fields[#].default`	string	-	Default value written to the Header when the specified field does not exist
`[#].fields[#].fixed`	string	-	Fixed value written to the Header
`[#].fields[#].remap`	dict	`null`	Perform mapping conversion on the extracted field value
`[#].fields[#].remap_default`	string	-	Default value when no corresponding mapping value exists during mapping conversion of the extracted field value If not specified, the original value is used If specified as `null`, this field is ignored
`[#].filter`	dict/string	`null`	Filter for matching data Supports Tag filtering and filterString filtering

Custom Global Tags Generation Function ID¶

Function ID format is {Script Set ID}__{Script ID}.{Function Name}

Function definition is as follows:

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 table 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 to the Header {#example}

Writing 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
    # Can be abbreviated if only 1 field
    fields: host

Example Data

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

Example Written Header

X-Global-Tags: host=web-001

Writing All Data to the Same Workspace¶

Omitting 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

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

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

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

Use custom function 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.guance.com?token=tkn_E-commerce Space Token"
    },
    {
      "rules": ["{ DF_WORKSPACE_UUID = 'wksp_123' }"],  // Match specified workspace
      "url": "https://backup.guance.com?token=tkn_Backup Space Token"
    },
    {
      "rules": ["*"],  // Default rule (must exist)
      "url": "https://default.guance.com?token=tkn_Default Space Token"
    }
  ]
}

Rule Syntax Description¶

For detailed operators, refer to Filter Operators.

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 like Session Replay, Profiling
Performance Impact: Each additional sharding field increases memory usage by about 5%

5. Impact of Global Tags¶

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 Merging 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 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: Use independent configuration blocks for different data categories (keyevent/logging/metric)
Concise Fields: Use no more than 3 sharding identifiers for a single data category
Avoid Conflicts: Use different naming for sharding fields of different categories

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 the default rule exists
Identifier Not Injected	1. Verify Func configuration syntax 2. Check if the field is of 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 Configuration¶

Global Tags Generation Rules¶

Custom Global Tags Generation Function ID¶

Generation Effect Verification¶

Writing Event Data to the Same Workspace¶

Writing All Data to the Same Workspace¶

Other Cases¶

3. Dataway Sinker Rule Configuration¶

Rule File Example (sinker.json)¶

Rule Syntax Description¶

4. Datakit End Configuration Instructions¶

Basic Configuration¶

Precautions¶

5. Impact of Global Tags¶

1. Global Tag Example¶

2. Sharding Identifier Merging Logic¶

Extended Explanation: Sharding Other Data Types¶

1. Custom Sharding Rules¶

2. General Principles¶

Troubleshooting¶

Common Issues¶

Diagnostic Commands¶

Is this page helpful? ×

Rule File Example (`sinker.json`)¶