Centralized coordination of data collection tasks from multiple sources

• US 9,807,192 B1
• Filed: 04/28/2017
• Issued: 10/31/2017
• Est. Priority Date: 07/29/2013
• Status: Active Grant

• Current Assignee: Splunk Inc.
• Original Assignee: Splunk Inc.
• Inventors: Fletcher, Tristan, Bingham, Brian
• Primary Examiner(s): Jacobs, Lashonda
• Application Number:
  US15/582,542
• Time in Patent Office: 186 Days
• CPC Class Codes: G06F 16/316 Indexing structures
• CPC Class Codes: G06F 16/334 Query execution G06F16/335 takes precedence
• CPC Class Codes: G06F 9/4881 Scheduling strategies for dispatcher, e.g. round robin, multi-level priority queues
• CPC Class Codes: G06F 9/5005 to service a request
• CPC Class Codes: G06F 9/5011 the resources being hardware resources other than CPUs, Servers and Terminals
• CPC Class Codes: G06F 9/5033 considering data affinity
• CPC Class Codes: G06F 9/505 considering the load
• CPC Class Codes: G06F 9/54 Interprogram communication
• CPC Class Codes: H04L 63/08 for authentication of entities cryptographic mechanisms or cryptographic arrangements for entity authentication H04L9/32
• CPC Class Codes: H04L 67/59 Providing operational support to end devices by off-loading in the network or by emulation, e.g. when they are unavailable

Abstract

A scheduler manages execution of a plurality of data-collection jobs, assigns individual jobs to specific forwarders in a set of forwarders, and generates and transmits tokens (e.g., pairs of data-collection tasks and target sources) to assigned forwarders. The forwarder uses the tokens, along with stored information applicable across jobs, to collect data from the target source and forward it onto an indexer for processing. For example, the indexer can then break a data stream into discrete events, extract a timestamp from each event and index (e.g., store) the event based on the timestamp. The scheduler can monitor forwarders'"'"' job performance, such that it can use the performance to influence subsequent job assignments. Thus, data-collection jobs can be efficiently assigned to and executed by a group of forwarders, where the group can potentially be diverse and dynamic in size.

Claims

1. A computer-implemented method comprising:

identifying a plurality of first jobs, each first job in the plurality of first jobs including collecting data from a source;

assigning each first job in the plurality of first jobs to a forwarder in a set of forwarders, each forwarder in the set of forwarders being configured to collect data from a source and to transmit the collected data to an indexer to be processed by the indexer and stored;

transmitting, for each first job in the plurality of first jobs, data indicative of the first job to the forwarder assigned to the first job;

receiving, from each of one or more forwarders in the set of forwarders, a communication indicative of a completion status of the assigned first job;

identifying a second job that includes collecting data;

selecting a forwarder of the set of forwarders to perform the second job, based on the received communication indicative of the completion status of the assigned first job; and

assigning the second job to the selected forwarder.

2. The method of claim 1, wherein the method is performed by a job scheduler implemented as a logically separate entity from the forwarders and the indexer.

3. The method of claim 1, wherein the plurality of first jobs includes a job to collect unstructured data from a first source using a first application programming interface (API) and further includes a job to collect unstructured data from a second source using a second API.

4. The method of claim 1, further comprising:

transmitting, to each forwarder in the set of forwarders, credential information that identifies secure information for use by the forwarder to communicate with the source.

5. The method of claim 1, further comprising:

estimating a load for each forwarder in the set of forwarders based on the received communication;

wherein the selection is made based on the estimated loads.

6. The method of claim 1, further comprising:

receiving an identification of a new forwarder, wherein the identification of the new forwarder is received after each first job in the plurality of first jobs is assigned but before the selection of a forwarder to perform the second job is made; and

adding the new forwarder to the set of forwarders, wherein selecting the forwarder of the set of forwarders comprises comparing a first value for a variable relating to an estimated current workload for the new forwarder to a second value for the variable for another forwarder in the set of forwarders.

7. The method of claim 1, wherein the data transmitted indicative of the first job for each first job in the plurality of first jobs includes an identification of the source from which data is to be collected.

8. The method of claim 1, wherein the data transmitted indicative of the first job for each first job in the plurality of first jobs includes an identification of an indexer to which the forwarder is to send the collected data, and wherein the indexer is configured to segment the data into a plurality of events, extract a timestamp from one event of the plurality of events, and storing the event based on the extracted timestamp.

9. The method of claim 1, further comprising:

prior to selecting the selected forwarder, determining a number of worker processes present on each forwarder in the set of forwarders.

10. The method of claim 1, wherein, for a first job in the plurality of first jobs, the data to be collected includes unstructured data.

11. The method of claim 1, wherein the second job and one first job in the plurality of first jobs pertain to data provided by a same source, and wherein the selection of the forwarder to perform the second job is made based on a bias towards assigning the second job and the one first job to a same forwarder.

12. A system comprising:

a communication interface through which the system can communicate with a remote entity via a network; and

one or more data processors coupled to the communication interface and individually or collectively programmed to perform operations including;

identifying a plurality of first jobs, wherein each first job in the plurality of first jobs includes collecting data from a source,

assigning each first job in the plurality of first jobs to a forwarder in a set of forwarders, each forwarder in the set of forwarders being configured to collect data from a source and transmit the collected data to an indexer to be processed by the indexer and stored,

transmitting, for each first job in the plurality of first jobs, data indicative of the first job to the forwarder assigned to the first job,

receiving a communication from each of one or more forwarders in the set of forwarders, wherein the communication is indicative of a completion status of the assigned first job,

identifying a second job, wherein the second job includes collecting data,

selecting a forwarder of the set of forwarders to perform the second job, based on the received communication indicative of the completion status of the assigned first job, and

assigning the second job to the selected forwarder.

13. The system of claim 12, wherein the plurality of first jobs includes ajob to collect unstructured data from a first source using a first application programming interface (API) and further includes a job to collect unstructured data from a second external source using a second API.

14. The system of claim 12, wherein one of the plurality of first jobs includes collecting the data from a host of a web service.

15. The system of claim 12, wherein the operations further comprise:

transmitting, to each forwarder in the set of forwarders, credential information that identifies secure information for use by the forwarder to communicate with the external source.

16. The system of claim 12, wherein the operations further comprise:

estimating a load for each forwarder in the set of forwarders based on the received communication;

wherein the selection is made based on the estimated loads.

17. The system of claim 12, wherein the operations further comprise:

receiving an identification of a new forwarder, wherein the identification of the new forwarder is received after each first job in the plurality of first jobs is assigned but before the selection of a forwarder to perform the second job is made; and

adding the new forwarder to the set of forwarders, wherein selecting the forwarder of the set of forwarders comprises comparing a first value for a variable relating to an estimated current workload for the new forwarder to a second value for the variable for another forwarder in the set of forwarders.

18. The system of claim 12, wherein the data transmitted indicative of the first job for each first job in the plurality of first jobs includes an identification of an indexer to which the forwarder is to send the collected data, and wherein the target indexer is configured to segment the data into a plurality of events, extract a timestamp from one event of the plurality of events, and storing the event based on the extracted timestamp.

19. The system of claim 12, wherein the operations further comprise:

prior to selecting the selected forwarder, determining a number of worker processes present on each forwarder in the set of forwarders.

20. The system of claim 12, wherein, for a first job in the plurality of first jobs, the data to be collected includes unstructured data.

21. The system of claim 12, wherein the second job and one first job in the plurality of first jobs pertain to data provided by a same source, and wherein the selection of the forwarder to perform the second job is made based on a bias towards assigning the second job and the one first job to a same forwarder.

22. A non-transitory machine-readable medium storing instructions, execution of which in a processing system causes one or more processors in the processing system to:

identify a plurality of first jobs, wherein each first job in the plurality of first jobs includes collecting data from a source;

assign each first job in the plurality of first jobs to a forwarder in a set of forwarders, each forwarder in the set of forwarders being configured to collect data from a source and transmit the collected data to an indexer to be processed by the indexer and stored;

transmit, for each first job in the plurality of first jobs, data indicative of the first job, wherein the data is transmitted to the forwarder assigned to the first job;

receive a communication from each of one or more forwarders in the set of forwarders, wherein the communication is indicative of a completion status of the assigned first job;

identify a second job that includes collecting data;

select a forwarder of the set of forwarders to perform the second job, based on the received communication indicative of the completion status of the assigned first job; and

assign the second job to the selected forwarder.

23. The non-transitory machine-readable medium of claim 22, wherein the plurality of first jobs includes a job to collect unstructured data from a first source using a first application programming interface (API) and further includes a job to collect unstructured data from a second source using a second API.

24. The non-transitory machine-readable medium of claim 22, wherein the instructions are further configured to cause the one or more data processors to:

transmitting, to each forwarder in the set of forwarders, credential information that identifies secure information for use by the forwarder to communicate with the external source.

25. The non-transitory machine-readable medium of claim 22, wherein the instructions are further configured to cause the one or more data processors to:

estimate a load for each forwarder in the set of forwarders based on the received communication;

wherein the selection is made based on the estimated loads.

26. The non-transitory machine-readable medium of claim 22, wherein the instructions are further configured to cause the one or more data processors to:

receive an identification of a new forwarder, wherein the identification of the new forwarder is received after each first job in the plurality of first jobs is assigned but before the selection of a forwarder to perform the second job is made; and

add the new forwarder to the set of forwarders, wherein selecting the forwarder of the set of forwarders comprises comparing a first value for a variable relating to an estimated current workload for the new forwarder to a second value for the variable for another forwarder in the set of forwarders.

27. The non-transitory machine-readable medium of claim 22, wherein the data transmitted indicative of the first job for each first job in the plurality of first jobs includes an identification of a target indexer to which the forwarder is to send the collected data, and wherein the target indexer is configured to segment the data into a plurality of events, extract a timestamp from one event of the plurality of events, and storing the event based on the extracted timestamp.

28. The non-transitory machine-readable medium of claim 22, wherein the instructions are further configured to cause the one or more data processors to:

prior to selecting the selected forwarder, determining a number of worker processes present on each forwarder in the set of forwarders.

29. The non-transitory machine-readable medium of claim 22, wherein, for a first job in the plurality of first jobs, the data to be collected includes unstructured data.

30. The non-transitory machine-readable medium of claim 22, wherein the second job and one first job in the plurality of first jobs pertain to data provided by a same source, and wherein the selection of the forwarder to perform the second job is made based on a bias towards assigning the second job and the one first job to a same forwarder.