[Auto Import] CSV format support

[ilyannn]

Release Notes

Automatic Import can now create integrations for logs in the CSV format. Owing to the maturity of log format support, we thus remove the verbiage about requiring the JSON/NDJSON format.

Summary

Added: the CSV feature

The issue is #194342

When the user adds a log sample whose format is recognized as CSV by the LLM, we now parse the samples and insert the csv processor into the generated pipeline.

If the header is present, we use it for the field names and add a drop processor that removes a header from the document stream by comparing the values to the header values.

If the header is missing, we ask the LLM to generate a list of column names, providing some context like package and data stream title.

Should the header or LLM suggestion provide unsuitable for a specific column, we use column1, column2 and so on as a fallback. To avoid duplicate column names, we can add postfixes like _2 as necessary.

If the format appears to be CSV, but the csv processor returns fails, we bubble up an error using the recently introduced ErrorThatHandlesItsOwnResponse class. We also provide the first example of passing the additional attributes of an error (in this case, the original CSV error) back to the client. The error message is composed on the client side.

Removed: supported formats message

The message that asks the user to upload the logs in JSON/NDJSON format is removed in this PR:

Refactoring

The refactoring makes the "→JSON" conversion process more uniform across different chains and centralizes processor definitions in .../server/util/processors.ts.

Log format chain now expects the LLM to follow the SamplesFormat when providing the information rather than an ad-hoc format.

When testing, the fail method is not supported in jest, so it is removed.

Examples

Postgres logs (original)

When using the original Postgres log file from the integrations repository we get the following error:

The reason is that the file is indeed not a correct CSV file, due to line breaks leading to unmatched quotes:

2021-01-04 01:07:04.364 UTC,"postgres","postgres",86,"172.24.0.1:45126",5ff26a0c.56,9,"idle",2021-01-04 01:06:20 UTC,3/8,0,LOG,00000,"statement: SELECT name FROM  (SELECT pg_catalog.lower(name) AS name FROM pg_catalog.pg_settings   WHERE context != 'internal'   UNION ALL SELECT 'all') ss  WHERE substring(name,1,3)='log'
LIMIT 1000",,,,,,,,,"psql","client backend"

Potsgres logs (fixed)

If we remove the line breaks and perform some additional fixes to the reserved fields, the integration ai_postgres_202410081032-1.0.0.zip is generated successfully, with LLM filling some of the columns names:

Example event

{
  "ai_postgres_202410081032": {
    "logs": {
      "backend_type": "postmaster",
      "column11": "0",
      "column4": "1",
      "column7": "7",
      "error_code": "00000",
      "session_start_time": "2021-01-04 01:06:13 UTC",
      "timestamp": "2021-01-04 01:07:20.001 UTC"
    }
  },
  "ecs": {
    "version": "8.11.0"
  },
  "event": {
    "category": [
      "database",
      "configuration"
    ],
    "created": "2021-01-04T01:07:20.001Z",
    "original": "2021-01-04 01:07:20.001 UTC,,,1,,5ff26a05.1,7,,2021-01-04 01:06:13 UTC,,0,LOG,00000,\"parameter \"\"log_min_duration_statement\"\" changed to \"\"0\"\"\",,,,,,,,,\"\",\"postmaster\"",
    "start": "2021-01-04T01:06:13.000Z",
    "type": [
      "info",
      "change"
    ]
  },
  "log": {
    "level": "LOG"
  },
  "message": "parameter \"log_min_duration_statement\" changed to \"0\"",
  "tags": [
    "preserve_original_event"
  ]
}

Recorded Future

This sample log file has a header. We generate the ai_rf_url_202410082213-1.0.0.zip with the following sample:

PanOS System Log

This sample log file does not have a header. We provide a "Format" string from the docs as part of the datastream title.

We produce the integration ai_panos_202410082244-1.0.0.zip:

Note the usage of separate FUTURE_USE_N columns.

Without the hints, the results (ai_panos_202410082250-1.0.0.zip) are less impressive:

PanOS Tunnel Inspection Log

• Log sample has only one line in it
• We provide information from the docs
• Generated integration: ai_panos_tunnel_202410132313-1.0.0.zip
• Generated event (compare to the manually developed integration):

Example event

        {
            "ai_panos_tunnel_202410132313": {
                "log": {
                    "A_Slice_Differentiator": "asd",
                    "A_Slice_Service_Type": "ast",
                    "Action_Flags": "user",
                    "Action_Source": "action",
                    "Application_Category": "cat",
                    "Application_Characteristic": "char",
                    "Application_Container": "con",
                    "Application_Risk": "4",
                    "Application_SaaS": "no",
                    "Application_Sanctioned_State": "no",
                    "Application_Subcategory": "app-sc",
                    "Application_Technology": "tech",
                    "Destination_External_Dynamic_List": "100",
                    "Destination_Location": "dst-loc",
                    "Destination_Zone": "d-zone",
                    "Device_Group_Hierarchy_Level_1": "0",
                    "Device_Group_Hierarchy_Level_2": "0",
                    "Device_Group_Hierarchy_Level_3": "0",
                    "Device_Group_Hierarchy_Level_4": "0",
                    "Dynamic_User_Group": "dug",
                    "FUTURE_USE": "1",
                    "FUTURE_USE_2": "2561",
                    "Flags": "0",
                    "Generated_Time": "2021/11/23 00:44:44",
                    "High_Resolution_Timestamp": "2021-11-23T00:44:44.930-08:00",
                    "Inbound_Interface": "inbound",
                    "Log_Action": "log",
                    "Maximum_Encapsulation": "20",
                    "Monitor_Tag_IMEI": "imei",
                    "PCAP_ID": "pcap",
                    "PDU_Session_ID": "100",
                    "Packets_Sent": "10",
                    "Parent_Session_ID": "1000",
                    "Parent_Start_Time": "1000",
                    "Receive_Time": "2021/11/23 00:44:44",
                    "Remote_User_ID": "100",
                    "Remote_User_IP": "81.2.69.192",
                    "Sequence_Number": "1000",
                    "Serial_Number": "1234567890",
                    "Sessions_Closed": "1000",
                    "Sessions_Created": "1000",
                    "Source_External_Dynamic_List": "100",
                    "Source_Location": "src-loc",
                    "Source_Zone": "s-zone",
                    "Start_Time": "2021-11-23T00:44:44.930-08:00",
                    "Strict_Check": "75",
                    "Subtype": "start",
                    "Tunnel": "1000",
                    "Tunnel_Fragment": "50",
                    "Tunnel_ID_IMSI": "imsi",
                    "Tunnel_Inspection_Rule": "rule1",
                    "Type": "START",
                    "Unknown_Protocol": "200",
                    "Virtual_System": "vsys",
                    "Virtual_System_Name": "vsys-name"
                }
            },
            "destination": {
                "bytes": "10",
                "geo": {
                    "city_name": "Changchun",
                    "continent_name": "Asia",
                    "country_iso_code": "CN",
                    "country_name": "China",
                    "location": {
                        "lat": 43.88,
                        "lon": 125.3228
                    },
                    "region_iso_code": "CN-22",
                    "region_name": "Jilin Sheng"
                },
                "ip": "175.16.199.1",
                "nat": {
                    "ip": "10.0.0.30",
                    "port": "9300"
                },
                "packets": "10",
                "port": "9550",
                "user": {
                    "name": "d-user"
                }
            },
            "ecs": {
                "version": "8.11.0"
            },
            "event": {
                "action": "action",
                "category": [
                    "network",
                    "authentication",
                    "session"
                ],
                "duration": "1234567890",
                "id": "id",
                "original": "1,2021/11/23 00:44:44,1234567890,START,start,2561,2021/11/23 00:44:44,10.0.0.10,175.16.199.1,10.0.0.20,10.0.0.30,rule,,d-user,app,vsys,s-zone,d-zone,inbound,outbound,log,,id,100,9000,9550,9200,9300,0,tcp,action,4,1000,user,src-loc,dst-loc,0,0,0,0,vsys-name,d-name,imsi,imei,1000,1000,1000,10,10,10,10,10,10,20,200,75,50,1000,1000,end,action,2021-11-23T00:44:44.930-08:00,1234567890,rule1,81.2.69.192,100,100,pcap,dug,100,100,2021-11-23T00:44:44.930-08:00,asd,ast,100,app-sc,cat,tech,4,char,con,no,no",
                "reason": "end",
                "sequence": "100",
                "severity": "4",
                "start": "2021-11-23T08:44:44.930Z",
                "type": [
                    "start",
                    "connection",
                    "info",
                    "end"
                ]
            },
            "host": {
                "name": "d-name"
            },
            "network": {
                "application": "app",
                "bytes": "10",
                "name": "outbound",
                "packets": "10",
                "transport": "tcp"
            },
            "related": {
                "hosts": [
                    "d-name"
                ],
                "ip": [
                    "10.0.0.30",
                    "175.16.199.1",
                    "10.0.0.20",
                    "10.0.0.10",
                    "81.2.69.192"
                ],
                "user": [
                    "d-user"
                ]
            },
            "rule": {
                "name": "rule",
                "uuid": "100"
            },
            "source": {
                "bytes": "10",
                "ip": "10.0.0.10",
                "nat": {
                    "ip": "10.0.0.20",
                    "port": "9200"
                },
                "port": "9000"
            },
            "tags": [
                "preserve_original_event"
            ]
        }

Todo

Completed items

• Unit or functional tests were updated or added to match the most common scenarios
• Strings are internationalized
• Upload an example generated for CSV
• Upload more examples

Possible follow-up

CSV feature:

• Also use package and data stream description to generate the columns; let the user know they can use this option to provide instructions to the LLM. → [Automatic Import] Explicit instructions to generate the CSV columns #196042
• If the CSV processor returns an error, provide a richer error description and an option to ignore it and continue the generation. → [Automatic Import] Rich error message for CSV parsing errors #196043

Log formats:

• Provide a new message to tell the user about the formats we support on the upload screen as well as in the error that is shown for unsupported formats. → [Automatic Import] Tell users about the log formats we support #196039
• If the LLM does not return a valid answer when asked about the log format, we currently default to unsupported. Instead we can ask again. → [Automatic Import] Retry if the LLM does not return a valid log format answer #196038

Bugs:

• When mapping to ECS fields, we do not ask LLM to map to @timestamp – a lot of our integrations don't have this field as a result. → [Automatic Import] Ensure the @timestamp field is present #196040
• When mapping the fields we don't convert them to the ECS field format, e.g. to long, which makes elastic-package test unhappy. → [Automatic Import] Conversion to target field type #196041

[elasticmachine]

Pinging @elastic/security-scalability (Team:Security-Scalability)

[ilyannn]

@elasticmachine merge upstream

[ilyannn]

@elasticmachine merge upstream

[elasticmachine]

merge conflict between base and head

[ilyannn]

@elasticmachine merge upstream

[ilyannn]

@elasticmachine merge upstream

[elasticmachine]

💔 Build Failed

• Buildkite Build
• Commit: b519093

Failed CI Steps

• Quick Checks

History

• 💔 Build #241897 failed 99e4ae3
• 💛 Build #241702 was flaky 2afe829
• 💔 Build #241425 failed 25d8009
• 💔 Build #241343 failed 85e1063
• 💚 Build #240154 succeeded c045198

cc @ilyannn

[bhapas]

LGTM

[kibanamachine]

Starting backport for target branches: 8.x

https://github.com/elastic/kibana/actions/runs/11325642463

[kibanamachine]

💚 All backports created successfully

Status	Branch	Result
✅	8.x

Note: Successful backport PRs will be merged automatically after passing CI.

Questions ?

Please refer to the Backport tool documentation