Create Ingest class that read from SplunkLoggings and processes logs

[zak-hassan]

Is your feature request related to a problem? Please describe.
Users may want to use splunklogs instead of elasticsearch. We should be able to support that.

Describe the solution you'd like
Get splunk python sdk and process logs from splunk.
https://dev.splunk.com/python

Is your feature request related to a problem? Please describe.
When we get predictions we want to write it to storage sink. To contribute a custom storage sink.

Step 1:
I would like you to extend this class

log-anomaly-detector/anomaly_detector/storage/storage_source.py

Lines 5 to 15 in 1e25ca8

	class StorageSource(metaclass=ABCMeta):
	"""Base class for storage implementations."""
	def __init__(self, configuration):
	"""Initialize storage."""
	self.config = configuration
	@abstractmethod
	def retrieve(self, storage_attribute):
	"""Retrieve data from storage and return them as a pandas dataframe."""
	raise NotImplementedError("Please implement the <retrieve method>")

Step 2:
create another function in this class.

log-anomaly-detector/anomaly_detector/storage/storage_catalog.py

Lines 50 to 54 in 1e25ca8

	@classmethod
	def _kafka_datasink_api(cls, config):
	"""Kafka data sink."""
	logging.info("save kafka datasink")
	return KafkaSink(config=config)

Step 3:
Add it to the catalog

log-anomaly-detector/anomaly_detector/storage/storage_catalog.py

Line 57 in 1e25ca8

'local.source': _localfile_datasource_api,

Additional context

Here is an example:

log-anomaly-detector/anomaly_detector/storage/local_storage.py

Lines 34 to 62 in 1e25ca8

	class LocalStorageDataSource(StorageSource, DataCleaner):
	"""Local storage Data source implementation."""
	NAME = "local.source"
	def __init__(self, configuration):
	"""Initialize local storage backend."""
	self.config = configuration
	def retrieve(self, storage_attribute: DefaultStorageAttribute):
	"""Retrieve data from local storage."""
	data = []
	_LOGGER.info("Reading from %s" % self.config.LS_INPUT_PATH)
	with open(self.config.LS_INPUT_PATH, "r") as fp:
	if self.config.LS_INPUT_PATH.endswith("json"):
	data = json.load(fp)
	else:
	# Here we are loading in data from common log format Columns [0]= timestamp [1]=severity [2]=msg
	for line in fp:
	message_field = " ".join(line.split(" ")[2:])
	message_field = message_field.rstrip("\n")
	data.append({"message": message_field})
	if storage_attribute.false_data is not None:
	data.extend(storage_attribute.false_data)
	data_set = json_normalize(data)
	_LOGGER.info("%d logs loaded", len(data_set))
	self._preprocess(data_set)
	return data_set, data

Any custom configurations go in this file:

log-anomaly-detector/anomaly_detector/config.py

Lines 103 to 107 in 1e25ca8

	# local test dataset
	LS_INPUT_PATH = ""
	# Name of local results data
	LS_OUTPUT_PATH = ""
	LS_OUTPUT_RWA_MODE = "w"

Describe alternatives you've considered
None
Additional context
N/A

[sesheta]

Issues go stale after 90d of inactivity.
Mark the issue as fresh with /remove-lifecycle stale.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with /close.

/lifecycle stale

[sesheta]

Stale issues rot after 30d of inactivity.
Mark the issue as fresh with /remove-lifecycle rotten.
Rotten issues close after an additional 30d of inactivity.

If this issue is safe to close now please do so with /close.

/lifecycle rotten

[sesheta]

Rotten issues close after 30d of inactivity.
Reopen the issue with /reopen.
Mark the issue as fresh with /remove-lifecycle rotten.

/close

[sesheta]

@sesheta: Closing this issue.

In response to this:

Rotten issues close after 30d of inactivity.
Reopen the issue with /reopen.
Mark the issue as fresh with /remove-lifecycle rotten.

/close

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.