Dynamic scheduling of tasks for collecting and processing data from external sources

• US 8,978,036 B2
• Filed: 07/29/2013
• Issued: 03/10/2015
• 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): An, Meng
• Assistant Examiner(s): Lee, James J
• Application Number:
  US13/953,568
• Publication Number:
  US 20130318536A1
• Time in Patent Office: 589 Days
• US Class Current: 718/104
• 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, by a scheduler, a plurality of jobs, wherein each job in the plurality of jobs comprises collecting data from one or more external sources;

job includes collecting data from an external source;

transmitting, at a direction of the scheduler, credentials to each forwarder in a set of multiple forwarders, wherein each forwarder stores the credentials;

generating, by the scheduler, a configuration token for each job, wherein the configuration token includes identification of data and a target source from which the data is to be collected;

selecting, by the scheduler, for each job a particular forwarder and assigning the job to the particular forwarder selected, wherein assigning the job to the particular forwarder takes into account information received from the particular forwarder on completion of previously assigned jobs;

transmitting, by the scheduler, the configuration token to the particular forwarder;

using, by the particular forwarder, the configuration token and the stored credentials in combination as needed to execute the job;

establishing, by the particular forwarder, communication with the target source using the credentials;

collecting, by the particular forwarder, the data from the target source using the configuration token and then forwarding the collected data to a particular indexer in a plurality of indexers to be indexed; and

receiving, by the scheduler, a communication from each particular forwarder wherein the communication is indicative of whether the assigned job has been completed.

2. The method of claim 1, wherein the plurality of first jobs includes a job to collect unstructured data from a first external 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.

3. The method of claim 1, wherein one of the plurality of jobs includes collecting the data from an external host of a web service.

4. The method of claim 1, further comprising:

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

wherein the assigning of jobs is made based on the estimated loads.

5. 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 job in the plurality of jobs is assigned but before the selection of a further forwarder to perform a further job is made; and

adding the new forwarder to the set of forwarders, wherein selecting the further forwarder of the set of forwarders comprises comparing an estimated current workload for the new forwarder to an estimated current workload for another forwarder in the set of forwarders.

6. The method of claim 1, wherein the configuration token further includes an identification of a particular indexer, of the plurality of indexers, to which the forwarder is to send the collected data.

7. The method of claim 1, further comprising:

prior to selecting the particular forwarder, determining a number of worker processes present on each forwarder in the set of forwarders, wherein selecting the particular forwarder is based at least in part on the number of worker processes present on each forwarder in the set of forwarders.

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

9. The method of claim 1, wherein a second job and one first job in the plurality of 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 first job to a same forwarder.

10. The method of claim 1, further comprising:

sending a first configuration token for a first job to the particular forwarder, the first configuration token comprising at least target configuration metadata for accessing a target resource;

determining, at the scheduler, that a second job also involves accessing the target resource;

responsive to determining that the second job also involves accessing the target resource, sending a second configuration token to the particular forwarder without resending the target configuration metadata.

11. The method of claim 1, further comprising:

accessing, at the scheduler, workload data that describes at least a collection task and a schedule of times at which the collection task is to be performed;

dividing, by the scheduler, the workload into at least a first job of the plurality of jobs and a second job.

12. The method of claim 1, further comprising:

assigning, by the scheduler, a first job to the particular forwarder;

wherein selecting the particular forwarder to perform a second job is based at least in part on the first job and the second job having both originated from a same workload.

13. The system of claim 1, further comprising:

maintaining capability data describing, for each forwarder in the set of forwarders, one or more handlers at the forwarder and one or more data collection tasks that each of the one or more handlers is configured to perform;

wherein assigning each job in the plurality of jobs is based at least partially on the capability data.

14. The method of claim 1,

wherein the scheduler executes at a first set of one or more computing device;

wherein the set of forwarders executes at a first set of one or more computing devices that includes at least one computing device other than the particular computing device;

wherein the plurality of indexers execute at a second set of one or more computing devices that is different than the first set of one or more computing devices.

15. The method of claim 1, further comprising:

executing, by a particular forwarder, a second job by establishing or validating a communication channel with a target resource identified by a particular configuration token received from the scheduler, collecting particular data from the target resource based on data collection configuration parameters described by the particular configuration token, formatting the particular data, forwarding the particular data to a particular indexer, and sending a particular communication to the scheduler that the second job is complete;

at the particular indexer, receiving the particular data, segmenting the particular data into a plurality of events, extracting timestamps from the plurality of events, storing the plurality of events based on the extracted timestamps, and

generating one or more indexes based on the plurality of events.

16. The method of claim 1, wherein assigning a second job to the particular forwarder is based at least in part on the received communication not including a communication that indicates that a first forwarder has completed a first job to which the first forwarder was assigned.

17. A system, comprising:

a plurality of data processors associated with a scheduler and a set of forwarders; and

a non-transitory computer-readable storage medium containing instructions which when executed on the plurality of data processors, cause the processors to perform operations including;

identifying, by a scheduler, a plurality of jobs, wherein each job in the plurality of jobs comprises collecting data from one or more external sources;

transmitting, at a direction of the scheduler, credentials to each forwarder in a set of multiple forwarders, wherein each forwarder stores the credentials;

generating, by the scheduler, a configuration token for each job, wherein the configuration token includes identification of data and a target source from which the data is to be collected;

selecting, by the scheduler, for each job a particular forwarder and assigning the job to the particular forwarder selected, wherein assigning the job to the particular forwarder takes into account information received from the particular forwarder on completion of previously assigned jobs;

transmitting, by the scheduler, the configuration token to the particular forwarder;

using, by the particular forwarder, the configuration token and the stored credentials in combination as needed to execute the job;

establishing, by the particular forwarder, communication with the target source using the credentials;

collecting, by the particular forwarder, the data from the target source using the configuration token and then forwarding the collected data to a particular indexer in a plurality of indexers to be indexed; and

receiving, by the scheduler, a communication from each particular forwarder wherein the communication is indicative of whether the assigned job has been completed.

18. The system of claim 17, wherein the plurality of jobs includes a job to collect unstructured data from a first external 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.

19. The system of claim 17, wherein one of the plurality of jobs includes collecting the data from an external host of a web service.

20. The system of claim 17, 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.

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

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

adding the new forwarder to the set of forwarders, wherein selecting the further forwarder of the set of forwarders comprises comparing an estimated current workload for the new forwarder to an estimated current workload for another forwarder in the set of forwarders.

22. The system of claim 17, wherein the configuration token further includes an identification of a particular indexer, of the plurality of indexers, to which the forwarder is to send the collected data.

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

prior to selecting the particular forwarder, determining a number of worker processes present on each forwarder in the set of forwarders, wherein selecting the particular forwarder is based at least in part on the number of worker processes present on each forwarder in the set of forwarders.

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

25. The system of claim 17, wherein a second job and one first job in the plurality of 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.

26. A computer-program product tangibly embodied in a non-transitory machine-readable storage medium, including instructions configured to cause a plurality of data processors associated with a scheduler and a set of forwarders to:

identify, by a scheduler, a plurality of jobs, wherein each job in the plurality of jobs comprises collecting data from one or more external sources;

transmit, at a direction of the scheduler, credentials to each forwarder in a set of multiple forwarders, wherein each forwarder stores the credentials;

generate, by the scheduler, a configuration token for each job, wherein the configuration token includes identification of data and a target source from which the data is to be collected;

select, by the scheduler, for each job a particular forwarder and assigning the job to the particular forwarder selected, wherein assigning the job to the particular forwarder takes into account information received from the particular forwarder on completion of previously assigned jobs;

transmit, by the scheduler, the configuration token to the particular forwarder;

use, by the particular forwarder, the configuration token and the stored credentials in combination as needed to execute the job;

establish, by the particular forwarder, communication with the target source using the credentials;

collect, by the particular forwarder, the data from the target source using the configuration token and then forwarding the collected data to a particular indexer in a plurality of indexers to be indexed; and

receive, by the scheduler, a communication from each particular forwarder, wherein the communication is indicative of whether the assigned job has been completed.