I have a number of small files, each of which maps to a single event. Since these files aren't actively added to (one-shot deals), I have been using batch mode to ingest them. I recently saw a problem during some testing. If I added the same file to the monitor directory a second time, the event in the file was ingested again and the search now shows multiple instances of the same event. I thought that Splunk was able to avoid duplicate entries by hashing the file and doing a CRC check. However, that does not seem to be working. The file that I have added has the same contents and the same name. The only thing different is the timestamp of the file since it was recently copied into the monitor directory. Here are the contents of my inputs.conf. [batch:///data/metadata] disabled = false move_policy = sinkhole sourcetype = meta I haved tried with and without crcSalt = <SOURCE> but it made no difference; the file is re-ingested every time I copy it into the batch monitor directory creating another duplicate event. Is there a way to avoid re-ingestion? Failing that, can I do something in search so that everyone only sees the most recent instance of the event? Thanks! ... View more

I have JSON data that I am ingesting. I would like to route the event to an index based on one of the JSON fields. I've seen examples that use REGEX, but I want to avoid hard coding the indexes since I will need to update multiple config files if I start getting new types of data. My JSON data includes the following section: ... "collection": { "date": "...", "source": <Canada | US | Mexico> }, ... I would like to have 3 seperate indexes, one for Canada, US, and Mexico. I would like to have the index determine dynamically based on the input. I've seen examples that suggest this is easy to do with REGEX, and I think I could do this as follows that way: indexes.conf: [index-Canada] ... [index-US] ... [index-Mexico] ... props.conf: [default] TRUNCATE = 0 INDEX_EXTRACTIONS = json TIMESTAMP_FIELDS = collection.date TRANSFORMS-SetIndex = setIndex-Canada, setIndex-US, setIndex-Mexico transforms.conf: [setIndex-Canada] REGEX = "source": "Canada" DEST_KEY = _MetaData::Index FORMAT = index-Canada [setIndex-US] REGEX = "source": "US" DEST_KEY = _MetaData::Index FORMAT = index-US [setIndex-Mexico] REGEX = "source": "Mexico" DEST_KEY = _MetaData::Index FORMAT = index-Mexico I think this will work. However, I would like to make it so that I don't have to hard code the transforms.conf for each index. One way is to do the following: props.conf: [default] TRUNCATE = 0 INDEX_EXTRACTIONS = json TIMESTAMP_FIELDS = collection.date TRANSFORMS-SetIndex = setIndex transforms.conf: [setIndex] REGEX = "source": "(.*)" DEST_KEY = _MetaData::Index FORMAT = index-$1 I have a couple questions about this: If the data has an index I haven't configured, can I somehow setup a fallback so that events that don't match a configured index are not lost? Can I use the SOURCE_KEY somehow to use the value of the JSON field instead of REGEX? I would rather use the JSON parsing ability of Splunk than my REGEX skills to make sure I am getting the right field. If somehow my REGEX shows up in the contents of the event later, I could get data routed to the wrong index. ... View more

For our solution, we need to index a number of events, but delete the events when they get too old. In our implementation, this is something like 6 months to a year. Each event references a file on a file system which also needs to be deleted when the event is deleted. The events that are indexed are JSON files. I've implemented something that works using a 'one-shot' search, but I would like some feedback to see if there are alternatives that might work better for my use case. I have a python script which connects to my splunk instance. I have a cron job scheduled for once a day which then executes the following with the Splunk API: import splunklib.client as client import json # Init connection to Splunk service = client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, username=SPLUNK_USER, password=SPLUNK_PASSWD) # Get all the events older than a year. kwargs_searcholder = {"latest_time" : "-1y"} search = "search index=custom" search_results = service.jobs.oneshot(search, **kwargs_searcholder) search_reader = search_results.ResultsReader(search_results) for item in search_reader: jsondata=json.loads(item['_raw']) id = jsondata['id'] # Delete the file on the file system delete_file(id) # Delete the event in splunk service.jobs.oneshot(search + " id=" + id + " | delete", **kwargs_deleteone) I have a few questions about this. Q1: is there a way to do this within splunk instead of creating a cron job? I need to do more than just run a search, I also need to delete a file on the file system. Q2: If the search returns a lot of records, then I am invoking the API once per record to perform the delete. Is there some way I can delete the same records I got in the search without specifying them all individually? I know I could just pipe the search through delete, but since this procedure takes a non-trivial amount of time, there will be events that are not old enough at the time of the first search, but are old enough at the time of the second. Then I would have files left on disk that are orphaned because the event is already deleted in Splunk. Q3: Is this a good candidate for a saved search? ... View more