msticpy.data.drivers.openobserve_driver module

OpenObserve Driver class.

class msticpy.data.drivers.openobserve_driver.OpenObserveDriver(**kwargs)

Bases: DriverBase

Driver to connect and query from OpenObserve.

Instantiate OpenObserve Driver.

add_query_filter(name, query_filter)

Add an expression to the query attach filter.

Parameters:

  • name (str)

  • query_filter (str | Iterable)

connect(connection_str=None, **kwargs)

Connect to OpenObserve via python-openobserve.

Parameters:

  • connection_str (str | None) – OpenObserve API url endpoint. default: https://localhost:5080

  • kwargs – Connection parameters can be supplied as keyword parameters.

Notes

Default configuration is read from the DataProviders/OpenObserve section of msticpyconfig.yaml, if available. There is not doing an http(s) connection, just filling underlying class properties.

property connected: bool

Return true if at least one connection has been made.

Returns:

True if a successful connection has been made.

Return type:

bool

Notes

This does not guarantee that the last data source connection was successful. It is a best effort to track whether the provider has made at least one successful authentication.

property driver_queries: Iterable[dict[str, Any]]

Return queries retrieved from the service after connecting.

Returns:

List of Dictionary of query_name, query_text. Name of container to add queries to.

Return type:

List[Dict[str, str]]

get_driver_property(name)

Return value or KeyError from driver properties.

Parameters:

name (str)

Return type:

Any

static get_http_timeout(**kwargs)

Get http timeout from settings or kwargs.

property instance: str | None

Return instance name, if one is set.

Returns:

The name of driver instance or None if the driver does not support multiple instances

Return type:

Optional[str]

property loaded: bool

Return true if the provider is loaded.

Returns:

True if the provider is loaded.

Return type:

bool

Notes

This is not relevant for some providers.

query(query, query_source=None, **kwargs)

Execute OpenObserve query and retrieve results.

Parameters:

  • query (str) – OpenObserve query to execute

  • query_source (QuerySource | None) – Not used.

  • days (int) – Search the past X days.

  • start (datetime) – A datetime() object representing the start of the search window. If used without end_time, the end of the search window is the current time.

  • start_time (datetime) – alias for start

  • end (datetime) – A datetime() object representing the end of the search window. If used without start_time, the search start will be the earliest time in the index.

  • end_time (datetime) – alias for end

  • timeZone (str) – timezone used for time range search

  • limit (int) – An integer describing the max number of search results to return.

  • verbosity (int) – Provide more verbose state. from 0 least verbose to 4 most one.

  • timeout (int) – timeout in seconds when gathering results

  • exporting (bool) – Export result to file.

  • export_path (str) – file path for exporte results.

  • time_columns (array[string]) – returning columns which format should be dataframe timestamp

  • numeric_columns (array[string]) – returning columns which format should be dataframe numeric

Returns:

Query results in a dataframe. or query response if an error.

Return type:

pd.DataFrame | Any

property query_attach_spec: dict[str, set[str]]

Parameters that determine whether a query is relevant for the driver.

query_usable(query_source)

Return True if query should be exposed for this driver.

Parameters:

query_source (QuerySource)

Return type:

bool

query_with_results(query, **kwargs)

Execute query string and return DataFrame of results.

Parameters:

query (str) – Query to execute against OpenObserve instance.

Returns:

A DataFrame (if successful) or the underlying provider result if an error occurs.

Return type:

tuple[pd.DataFrame, Any]

property schema: dict[str, dict]

Return current data schema of connection.

Returns:

Data schema of current connection.

Return type:

Dict[str, Dict]

property service_queries: tuple[dict[str, str], str]

Return queries retrieved from the service after connecting.

Returns:

Dictionary of query_name, query_text. Name of container to add queries to.

Return type:

Tuple[Dict[str, str], str]

set_driver_property(name, value)

Set an item in driver properties.

Parameters:

  • name (str)

  • value (Any)