Processing engine and Python plugins

Use the Processing Engine in InfluxDB 3 Core to extend your database with custom Python code. Trigger your code on write, on a schedule, or on demand to automate workflows, transform data, and create API endpoints.

What is the Processing Engine?

The Processing Engine is an embedded Python virtual machine that runs inside your InfluxDB 3 Core database. You configure triggers to run your Python plugin code in response to:

• Data writes - Process and transform data as it enters the database
• Scheduled events - Run code at defined intervals or specific times
• HTTP requests - Expose custom API endpoints that execute your code

You can use the Processing Engine’s in-memory cache to manage state between executions and build stateful applications directly in your database.

This guide walks you through setting up the Processing Engine, creating your first plugin, and configuring triggers that execute your code on specific events.

Before you begin

Ensure you have:

• A working InfluxDB 3 Core instance
• Access to command line
• Python installed if you’re writing your own plugin
• Basic knowledge of the InfluxDB CLI

Once you have all the prerequisites in place, follow these steps to implement the Processing Engine for your data automation needs.

• Set up the Processing Engine
• Add a Processing Engine plugin
  • Upload plugins from local machine
  • Update existing plugins
  • View loaded plugins
• Create a trigger
• Manage plugin dependencies
• Plugin security

Set up the Processing Engine

The Processing Engine activates when --plugin-dir or INFLUXDB3_PLUGIN_DIR is configured.

Default behavior by deployment type

If you installed InfluxDB 3 Core using Docker or a DEB/RPM package, the Processing Engine is already enabled—skip to Add a Processing Engine plugin. To disable the Processing Engine, see Enable and disable the Processing Engine.

Enable the Processing Engine manually

To activate the Processing Engine when running from a binary or source build, start your InfluxDB 3 Core server with the --plugin-dir flag. This flag tells InfluxDB where to load your plugin files.

your-install-location/
├── influxdb3
└── python/

influxdb3 serve \
  --
NODE_ID
 \
  --object-store 
OBJECT_STORE_TYPE
 \
  --plugin-dir 
PLUGIN_DIR

In the example above, replace the following:

• NODE_ID: Unique identifier for your instance
• OBJECT_STORE_TYPE: Type of object store (for example, file or s3)
• PLUGIN_DIR: Absolute path to the directory where plugin files are stored. Store all plugin files in this directory or its subdirectories.

Configure distributed environments

When running InfluxDB 3 Core in a distributed setup, follow these steps to configure the Processing Engine:

1. Decide where each plugin should run
   • Data processing plugins, such as WAL plugins, run on ingester nodes
   • HTTP-triggered plugins run on nodes handling API requests
   • Scheduled plugins can run on any configured node
2. Enable plugins on the correct instance
3. Maintain identical plugin files across all instances where plugins run
   • Use shared storage or file synchronization tools to keep plugins consistent

Add a Processing Engine plugin

A plugin is a Python script that defines a function with a trigger-compatible (trigger spec) signature. When the specified event occurs, InfluxDB runs the plugin.

Choose a plugin strategy

You have two main options for adding plugins to your InfluxDB instance:

• Use example plugins - Get started with prebuilt plugins
• Create a custom plugin - Build your own for specialized use cases

Use example plugins

InfluxData maintains a repository of official and community plugins that you can use immediately in your Processing Engine setup.

Browse the plugin library to find examples and InfluxData official plugins for:

• Data transformation: Process and transform incoming data
• Alerting: Send notifications based on data thresholds
• Aggregation: Calculate statistics on time series data
• Integration: Connect to external services and APIs
• System monitoring: Track resource usage and health metrics

For community contributions, see the influxdb3_plugins repository on GitHub.

Add example plugins

You have two options for using plugins from the repository:

Option 1: Copy plugins locally

Clone the influxdata/influxdb3_plugins repository and copy plugins to your configured plugin directory:

# Clone the repository
git clone https://github.com/influxdata/influxdb3_plugins.git
   
# Copy a plugin to your configured plugin directory
cp influxdb3_plugins/influxdata/system_metrics/system_metrics.py /path/to/plugins/

Option 2: Reference plugins directly from GitHub

Skip downloading plugins by referencing them directly from GitHub using the gh: prefix:

# Create a trigger using a plugin from GitHub
influxdb3 create trigger \
  --trigger-spec "every:1m" \
  --path "gh:influxdata/system_metrics/system_metrics.py" \
  --database my_database \
  system_metrics

This approach:

• Ensures you’re using the latest version
• Simplifies updates and maintenance
• Reduces local storage requirements

Syntax:

gh:<path-to-plugin-file>

The path after gh: is appended to the configured --plugin-repo.

By default, gh:-prefixed plugins resolve against the official influxdata/influxdb3_plugins repository at https://raw.githubusercontent.com/influxdata/influxdb3_plugins/main/. For example, gh:examples/wal_plugin/wal_plugin.py resolves to:

https://raw.githubusercontent.com/influxdata/influxdb3_plugins/main/examples/wal_plugin/wal_plugin.py

How gh: plugin resolution works:

1. InfluxDB 3 Core detects the gh: prefix in the plugin path.

2. It strips the prefix and appends the remaining path to the configured plugin repository URL.

3. An HTTP GET request fetches the plugin source code.

4. If the fetch succeeds (HTTP 2xx), InfluxDB 3 Core validates the plugin and creates the trigger.

5. If the fetch fails, the command returns an error with the HTTP status code and URL–for example:

   error fetching plugin from repository: 404 Not Found https://raw.githubusercontent.com/influxdata/influxdb3_plugins/main/not_found.py

InfluxDB 3 Core fetches the plugin at trigger creation time (to validate it), and again each time the trigger starts–for example, on server startup or when you re-enable a disabled trigger. Unlike local plugins, GitHub plugins aren’t automatically reloaded when the source changes–disable and re-enable the trigger to fetch updates. Only single-file plugins are supported through the gh: prefix; multi-file plugin directories must be uploaded locally (see Upload plugins from local machine).

Option 3: Use a custom plugin repository

For organizations that maintain their own plugin repositories or need to use private/internal plugins, configure a custom plugin repository URL:

# Start the server with a custom plugin repository
influxdb3 serve \
  --node-id node0 \
  --object-store file \
  --data-dir ~/.influxdb3 \
  --plugin-dir ~/.plugins \
  --plugin-repo "https://internal.company.com/influxdb-plugins/"

Then reference plugins from your custom repository using the gh: prefix:

# Fetches from: https://internal.company.com/influxdb-plugins/myorg/custom_plugin.py
influxdb3 create trigger \
  --trigger-spec "every:5m" \
  --path "gh:myorg/custom_plugin.py" \
  --database my_database \
  custom_trigger

Use cases for custom repositories:

• Private plugins: Host proprietary plugins not suitable for public repositories
• Air-gapped environments: Use internal mirrors when external internet access is restricted
• Development and staging: Test plugins from development branches before production deployment
• Compliance requirements: Meet data governance policies requiring internal hosting

The --plugin-repo option accepts any HTTP/HTTPS URL that serves raw plugin files. You can also set it with the INFLUXDB3_PLUGIN_REPO environment variable. See the plugin-repo configuration option for more details.

Plugins have various functions such as:

• Receive plugin-specific arguments (such as written data, call time, or an HTTP request)
• Access keyword arguments (as args) passed from trigger arguments configurations
• Access the influxdb3_local shared API to write data, query data, and managing state between executions

For more information about available functions, arguments, and how plugins interact with InfluxDB, see how to Extend plugins.

Create a custom plugin

To build custom functionality, you can create your own Processing Engine plugin.

Prerequisites

Before you begin, make sure:

• The Processing Engine is enabled on your InfluxDB 3 Core instance.
• You’ve configured the --plugin-dir where plugin files are stored.
• You have access to that plugin directory.

Steps to create a plugin:

• Choose your plugin type
• Create your plugin file
• Next Steps

Choose your plugin type

Choose a plugin type based on your automation goals:

Create your plugin file

Plugins now support both single-file and multifile architectures:

Single-file plugins:

• Create a .py file in your plugins directory
• Add the appropriate function signature based on your chosen plugin type
• Write your processing logic inside the function

Multifile plugins:

• Create a directory in your plugins directory
• Add an __init__.py file as the entry point (required)
• Organize supporting modules in additional .py files
• Import and use modules within your plugin code

Example multifile plugin structure

my_plugin/
├── __init__.py       # Required - entry point with trigger function
├── utils.py          # Supporting module
├── processors.py     # Data processing functions
└── config.py         # Configuration helpers

The __init__.py file must contain your trigger function:

# my_plugin/__init__.py
from .processors import process_data
from .config import get_settings

def process_writes(influxdb3_local, table_batches, args=None):
    settings = get_settings()
    for table_batch in table_batches:
        process_data(influxdb3_local, table_batch, settings)

Supporting modules can contain helper functions:

# my_plugin/processors.py
def process_data(influxdb3_local, table_batch, settings):
    # Processing logic here
    pass

After writing your plugin, create a trigger to connect it to a database event and define when it runs.

Create a data write plugin

Use a data write plugin to process data as it’s written to the database. These plugins use table: or all_tables: trigger specifications. Ideal use cases include:

• Data transformation and enrichment
• Alerting on incoming values
• Creating derived metrics

def process_writes(influxdb3_local, table_batches, args=None):
    # Process data as it's written to the database
    for table_batch in table_batches:
        table_name = table_batch["table_name"]
        rows = table_batch["rows"]
        
        # Log information about the write
        influxdb3_local.info(f"Processing {len(rows)} rows from {table_name}")
        
        # Write derived data back to the database
        line = LineBuilder("processed_data")
        line.tag("source_table", table_name)
        line