[axonius][user] Add Axonius User datastream by muskan-agarwal26 · Pull Request #16142 · elastic/integrations

muskan-agarwal26 · 2025-11-27T09:27:48Z

Proposed commit message

The initial release includes user data stream and associated dashboard.

Axonius fields are mapped to their corresponding ECS fields where possible.

Test samples were derived from live data samples, which were subsequently
sanitized.

Checklist

I have reviewed tips for building integrations and this pull request is aligned with them.
I have verified that all data streams collect metrics or logs.
I have added an entry to my package's changelog.yml file.
I have verified that Kibana version constraints are current according to guidelines.
I have verified that any added dashboard complies with Kibana's Dashboard good practices

How to test this PR locally

To test the axonius package:

Clone integrations repo.
Install elastic package locally.
Start elastic stack using elastic-package.
Move to integrations/packages/axonius directory.
Run the following command to run tests.

elastic-package test

2025/11/26 16:49:31  INFO New version is available - v0.117.1. Download from: https://github.com/elastic/elastic-package/releases/tag/v0.117.1
Run asset tests for the package
2025/11/26 16:49:31  INFO License text found in "/root/GITHUB/integrations/LICENSE.txt" will be included in package
--- Test results for package: axonius - START ---
╭─────────┬─────────────┬───────────┬──────────────────────────────────────────────────────────────────┬────────┬──────────────╮
│ PACKAGE │ DATA STREAM │ TEST TYPE │ TEST NAME                                                        │ RESULT │ TIME ELAPSED │
├─────────┼─────────────┼───────────┼──────────────────────────────────────────────────────────────────┼────────┼──────────────┤
│ axonius │             │ asset     │ dashboard axonius-cc6c1878-aa15-4765-a4d9-76eca1cc4078 is loaded │ PASS   │      1.961µs │
│ axonius │             │ asset     │ search axonius-723a9fbe-5e79-4c85-b6c8-486aa02fd95c is loaded    │ PASS   │        243ns │
│ axonius │ user        │ asset     │ index_template logs-axonius.user is loaded                       │ PASS   │        238ns │
│ axonius │ user        │ asset     │ ingest_pipeline logs-axonius.user-0.1.0 is loaded                │ PASS   │        209ns │
╰─────────┴─────────────┴───────────┴──────────────────────────────────────────────────────────────────┴────────┴──────────────╯
--- Test results for package: axonius - END   ---
Done
Run pipeline tests for the package
--- Test results for package: axonius - START ---
╭─────────┬─────────────┬───────────┬──────────────────────────────────────────┬────────┬──────────────╮
│ PACKAGE │ DATA STREAM │ TEST TYPE │ TEST NAME                                │ RESULT │ TIME ELAPSED │
├─────────┼─────────────┼───────────┼──────────────────────────────────────────┼────────┼──────────────┤
│ axonius │ user        │ pipeline  │ (ingest pipeline warnings test-user.log) │ PASS   │ 748.512078ms │
│ axonius │ user        │ pipeline  │ test-user.log                            │ PASS   │ 237.300595ms │
╰─────────┴─────────────┴───────────┴──────────────────────────────────────────┴────────┴──────────────╯
--- Test results for package: axonius - END   ---
Done
Run policy tests for the package
--- Test results for package: axonius - START ---
No test results
--- Test results for package: axonius - END   ---
Done
Run static tests for the package
--- Test results for package: axonius - START ---
╭─────────┬─────────────┬───────────┬──────────────────────────┬────────┬──────────────╮
│ PACKAGE │ DATA STREAM │ TEST TYPE │ TEST NAME                │ RESULT │ TIME ELAPSED │
├─────────┼─────────────┼───────────┼──────────────────────────┼────────┼──────────────┤
│ axonius │ user        │ static    │ Verify sample_event.json │ PASS   │ 207.716458ms │
╰─────────┴─────────────┴───────────┴──────────────────────────┴────────┴──────────────╯
--- Test results for package: axonius - END   ---
Done
Run system tests for the package
2025/11/26 16:49:37  INFO Installing package...
2025/11/26 16:49:37  INFO License text found in "/root/GITHUB/integrations/LICENSE.txt" will be included in package
2025/11/26 16:49:49  INFO Running test for data_stream "user" with configuration 'default'
2025/11/26 16:49:57  INFO Setting up independent Elastic Agent...
2025/11/26 16:50:09  INFO Setting up service...
2025/11/26 16:50:30  INFO Validating test case...
2025/11/26 16:50:31  INFO Tearing down service...
2025/11/26 16:50:32  INFO Write container logs to file: /root/GITHUB/integrations/build/container-logs/axonius-1764156032894361232.log
2025/11/26 16:50:34  INFO Tearing down agent...
2025/11/26 16:50:34  INFO Write container logs to file: /root/GITHUB/integrations/build/container-logs/elastic-agent-1764156034804125811.log
2025/11/26 16:50:53  INFO Uninstalling package...
--- Test results for package: axonius - START ---
╭─────────┬─────────────┬───────────┬───────────┬────────┬───────────────╮
│ PACKAGE │ DATA STREAM │ TEST TYPE │ TEST NAME │ RESULT │  TIME ELAPSED │
├─────────┼─────────────┼───────────┼──��────────┼────────┼───────────────┤
│ axonius │ user        │ system    │ default   │ PASS   │ 42.087523671s │
╰─────────┴─────────────┴───────────┴───────────┴────────┴───────────────╯
--- Test results for package: axonius - END   ---
Done

Screenshots

Go Code for Ingest Pipeline Generation

The user data stream pipeline is generated using Go code built on top of the Dispear library.
Below is the code used for generating the pipeline logic:

package main

import (
	"fmt"
	"strings"

	. "github.com/efd6/dispear"
)

const (
	ECSVersion = "9.2.0"
	PkgRoot    = "axonius.user"
)
const errorFormat = "Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}"

// removeErrorHandler generates a series of Renders that first remove the given field
// from the document and then append a custom error message to the 'error.message' field.
// This is typically used to handle errors in an Ingest Pipeline by removing a field that
// caused an issue and appending a formatted error message to the document.
func removeErrorHandler(f string) []Renderer {
	return []Renderer{
		REMOVE(f),
		APPEND("error.message", errorFormat),
	}
}

// safeNavigateAndCheck converts a dot-separated field path to a safe navigation string.
//
// Example:
// "parent.child.grandchild" -> "ctx?.parent?.child?.grandchild"
func safeNavigateAndCheck(field string) string {
	parts := strings.Split(field, ".")
	condition := "ctx"
	for i, part := range parts {
		if i > 0 { // Skip the first part which is already included in the condition
			condition += fmt.Sprintf("?.%s", part)
		} else {
			condition += fmt.Sprintf(".%s", part)
		}
	}
	return condition
}

func main() {

	// Initial processors of pipeline

	DESCRIPTION("Pipeline for processing user logs.")

	SET("ecs.version").VALUE(ECSVersion)

	TERMINATE("data collection error").
		IF("ctx.error?.message != null && ctx.message == null && ctx.event?.original == null").
		DESCRIPTION("error message set and no data to process.")

	BLANK()
	BLANK().COMMENT("remove agentless metadata")

	REMOVE(
		"organization",
		"division",
		"team",
	).
		IF("ctx.organization instanceof String && ctx.division instanceof String && ctx.team instanceof String").
		IGNORE_MISSING(true).
		TAG("remove_agentless_tags").
		DESCRIPTION("Removes the fields added by Agentless as metadata, as they can collide with ECS fields.")

	BLANK()
	BLANK().COMMENT("parse the event JSON")

	RENAME("message", "event.original").
		IF("ctx.event?.original == null").
		DESCRIPTION("Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.").
		IGNORE_MISSING(true)

	REMOVE("message").
		TAG("remove_message").
		IF("ctx.event?.original != null").
		DESCRIPTION("The `message` field is no longer required if the document has an `event.original` field.").
		IGNORE_MISSING(true)

	JSON(PkgRoot, "event.original")

	// Add fingerprint

	BLANK()
	BLANK().COMMENT("Add fingerprint")

	FINGERPRINT("_id", "axonius.user.uuid").IGNORE_MISSING(true)

	// Setting event.* fields

	BLANK()
	BLANK().COMMENT("Set event.* fields")

	SET("event.kind").VALUE("event")

	// Use Date processors

	BLANK()
	BLANK().COMMENT("Date processors")

	for _, field := range []string{
		"axonius.user.last_updated",
		"axonius.user.last_login",
	} {
		DATE(field, field, "EEE, dd MMM yyyy HH:mm:ss 'GMT'", "yyyy-MM-dd").
			IF(safeNavigateAndCheck(field) + " != null" + " && " + "ctx." + field + " != ''").
			ON_FAILURE(removeErrorHandler(field)...)
	}
	// Set ECS Mapping

	BLANK()
	BLANK().COMMENT("Map custom fields to corresponding ECS and related fields.")

	// Map ECS mapping for top-level fields

	for _, mapping := range []struct {
		ecsField, customField string
	}{
		{ecsField: "user.email", customField: "axonius.user.email"},
		{ecsField: "user.name", customField: "axonius.user.user_name"},
		{ecsField: "user.id", customField: "axonius.user.uuid"},
	} {
		SET(mapping.ecsField).COPY_FROM(mapping.customField).IGNORE_EMPTY(true)
	}

	for _, mapping := range []struct {
		ecsField, customField string
	}{
		{ecsField: "user.roles", customField: "axonius.user.role_name"},
	} {
		APPEND(mapping.ecsField, "{{{"+mapping.customField+"}}}").
			IF(safeNavigateAndCheck(mapping.customField) + " != null").
			ALLOW_DUPLICATES(false)
	}

	// Set user.full_name from user.first_name and axonius.user.first_name and axonius.user.last_name

	SET("user.full_name").
		VALUE("{{{axonius.user.first_name}}} {{{axonius.user.last_name}}}").
		IF("ctx.axonius?.user?.first_name != null && ctx.axonius?.user?.last_name != null")

	SET("user.email").
		COPY_FROM("user.name").
		IF("ctx.user?.email == null && ctx.user?.name != null && ctx.user?.name.contains('@')")

	DISSECT("user.email", "%{}@%{user.domain}").
		IF("ctx.user?.email != null && ctx.user.email.contains('@')")

	// Map related mapppings
	for _, mapping := range []struct {
		ecsField, customField string
	}{
		{ecsField: "related.user", customField: "axonius.user.first_name"},
		{ecsField: "related.user", customField: "axonius.user.last_name"},
		{ecsField: "related.user", customField: "axonius.user.email"},
		{ecsField: "related.user", customField: "axonius.user.user_name"},
		{ecsField: "related.user", customField: "axonius.user.uuid"},
	} {
		APPEND(mapping.ecsField, "{{{"+mapping.customField+"}}}").
			IF(safeNavigateAndCheck(mapping.customField) + " != null").
			ALLOW_DUPLICATES(false)
	}

	// Remove Duplicate Fields.

	BLANK()
	BLANK().COMMENT("Remove duplicate custom fields if preserve_duplicate_custom_fields are not enabled")

	REMOVE(
		"axonius.user.email",
		"axonius.user.user_name",
		"axonius.user.uuid",
	).
		IF("ctx.tags == null || !ctx.tags.contains('preserve_duplicate_custom_fields')").
		TAG("remove_custom_duplicate_fields").
		IGNORE_MISSING(true)

	// Clean up script

	BLANK()
	BLANK().COMMENT("Cleanup")

	SCRIPT().
		TAG("script_to_drop_null_values").
		DESCRIPTION("This script processor iterates over the whole document to remove fields with null values.").
		LANG("painless").
		SOURCE(`
		void handleMap(Map map) {
		map.values().removeIf(v -> {
			if (v instanceof Map) {
			handleMap(v);
			} else if (v instanceof List) {
			handleList(v);
			}
			return v == null || v == '' || (v instanceof Map && v.size() == 0) || (v instanceof List && v.size() == 0)
		});
		}
		void handleList(List list) {
		list.removeIf(v -> {
			if (v instanceof Map) {
			handleMap(v);
			} else if (v instanceof List) {
			handleList(v);
			}
			return v == null || v == '' || (v instanceof Map && v.size() == 0) || (v instanceof List && v.size() == 0)
		});
		}
		handleMap(ctx);
		`)

	// Set and Append processor on last

	SET("event.kind").VALUE("pipeline_error").IF("ctx.error?.message != null")
	APPEND("tags", "preserve_original_event").
		IF("ctx.error?.message != null").
		ALLOW_DUPLICATES(false)

	// Global on failure processor

	ON_FAILURE(
		APPEND("error.message", errorFormat),
		SET("event.kind").VALUE("pipeline_error"),
		APPEND("tags", "preserve_original_event").
			ALLOW_DUPLICATES(false),
	)

	// Generate the pipeline

	Generate()
}

efd6 · 2025-11-27T09:28:51Z

packages/axonius/changelog.yml

+  changes:
+    - description: Initial release.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1


Suggested change

link: https://github.com/elastic/integrations/pull/1

link: https://github.com/elastic/integrations/pull/16142

packages/axonius/_dev/build/docs/README.md

kcreddy · 2026-01-02T11:04:29Z

packages/axonius/elasticsearch/transform/latest_user/transform.yml

+# Use of "*" to use all namespaces defined.
+source:
+  index:
+    - "logs-axonius.user-*"


Permissions are added in elastic/elasticsearch#140046 ✅

packages/axonius/manifest.yml

packages/axonius/data_stream/user/_dev/test/pipeline/test-user.log-expected.json

muskan-agarwal26 requested a review from a team as a code owner November 27, 2025 09:27

efd6 reviewed Nov 27, 2025

View reviewed changes

muskan-agarwal26 marked this pull request as draft December 2, 2025 07:13

muskan-agarwal26 marked this pull request as ready for review December 31, 2025 12:32

muskan-agarwal26 requested a review from a team as a code owner December 31, 2025 12:32

muskan-crest added 4 commits January 1, 2026 12:24

Initial release of User data-stream in axonius

dfe2d6b

Update changelog entry

3a3a586

Added transform pipeline and ilm policy

a38f348

Added role permission note in readme

737c3a3

muskan-agarwal26 force-pushed the datastream-user branch from 924725e to 737c3a3 Compare January 1, 2026 09:07

Updated manifest versions as ILM Policy rely with these versions.

9a88f83

ShourieG assigned muskan-agarwal26 Jan 2, 2026

kcreddy self-requested a review January 2, 2026 11:06

kcreddy reviewed Jan 2, 2026

View reviewed changes

Resolved comments provided by @kcreddy

08390ac

muskan-agarwal26 requested a review from kcreddy January 5, 2026 09:41

kcreddy approved these changes Jan 6, 2026

View reviewed changes

kcreddy merged commit 4fee8da into elastic:feature/axonius-0.1.0 Jan 6, 2026
8 checks passed

andrewkroh removed the New Integration Issue or pull request for creating a new integration package. label Jan 8, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[axonius][user] Add Axonius User datastream#16142

[axonius][user] Add Axonius User datastream#16142
kcreddy merged 6 commits intoelastic:feature/axonius-0.1.0from
muskan-agarwal26:datastream-user

muskan-agarwal26 commented Nov 27, 2025 •

edited

Loading

efd6 Nov 27, 2025

Uh oh!

Uh oh!

kcreddy Jan 2, 2026

Uh oh!

Uh oh!

Uh oh!

Labels

5 participants

	link: https://github.com/elastic/integrations/pull/1
	link: https://github.com/elastic/integrations/pull/16142

Conversation

muskan-agarwal26 commented Nov 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Proposed commit message

Checklist

How to test this PR locally

Screenshots

Go Code for Ingest Pipeline Generation

efd6 Nov 27, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kcreddy Jan 2, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Labels

5 participants

muskan-agarwal26 commented Nov 27, 2025 •

edited

Loading