RING ARCHITECTURE-BASED WORKLOAD DISTRIBUTION IN A MICROSERVICE COMPUTING ENVIRONMENT

Techniques are disclosed for managing workloads in data processing systems. For example, a method determines a set of containers for processing a given workload of data records, wherein each container of the set of containers is configured to process a given subset of the data records. The method causes deployment of the set of containers in a ring configuration to process the given workload of data records such that at least one of the containers in the ring configuration automatically processes one or more data records of a subset of data records assigned to another container in the ring configuration that becomes inactive.

FIELD

The field relates generally to information processing systems, and more particularly to workload management in such information processing systems.

BACKGROUND

Microservices are the predominant approach in the modern development of software (e.g., application programs, or more simply, applications) across a wide variety of computing platforms such as, but not limited to, a cloud computing platform, a private computing platform, a hybrid (cloud/private) computing platform, an edge computing platform, etc. A microservice architecture manages an application as a collection of services. As such, development of an application can be accomplished in a flexible and scalable manner.

Initially, microservices were used in application programming interface (API) environments where synchronous/asynchronous request calls occur (e.g., web applications). However, microservices are now used in Docker and Kubernetes container environments, as well as for batch processing in data pipeline and other data processing architectures. Typically, in a data pipeline system, there are multiple datastores (stores) where data for a given job (workload) is read from a source datastore, transformed and (possibly, as needed/desired) stored in an intermediate datastore, and then ultimately loaded onto a destination datastore. Then, another given job is similarly processed from the source datastore through to the destination datastore.

In batch processing (i.e., wherein a job/workload is scheduled and runs in a specific interval), parallel threads (processors) are typically enabled to execute the data faster in both virtual machine and physical server environments. However, in parallel processing, there can be use cases where different threads process the same data causing duplicate entries or corrupted data in the destination datastore.

SUMMARY

Illustrative embodiments provide improved techniques for managing workloads in data processing systems.

For example, in an illustrative embodiment, a method comprises the following steps. The method determines a set of containers for processing a given workload of data records, wherein each container of the set of containers is configured to process a given subset of the data records. The method causes deployment of the set of containers in a ring configuration to process the given workload of data records such that at least one of the containers in the ring configuration automatically processes one or more data records of a subset of data records assigned to another container in the ring configuration that becomes inactive.

Further illustrative embodiments are provided in the form of a non-transitory computer-readable storage medium having embodied therein executable program code that when executed by a processor causes the processor to perform the above steps. Still further illustrative embodiments comprise an apparatus with a processor and a memory configured to perform the above steps.

Advantageously, illustrative embodiments distribute microservice containers in a ring architecture (ring model) to execute records in a given workload in a given direction around the ring architecture such that, if one microservice container fails, the next microservice container in the ring architecture automatically processes records of the failed microservice container.

DETAILED DESCRIPTION

As mentioned, batch processing systems have been used to process workloads associated with microservices or other job execution applications. However, in batch processing systems that use parallel processing, there can be use cases where different threads process the same data causing duplicate entries or corrupted data in the destination datastore. So, typically, a different data set (block) is assigned for each thread for processing.

For example, assume 60,000 (60K) records are to be processed, and a system instantiates (spins) three parallel threads and assigns the first 20K records (block comprising records 1 to 20K) to the first thread, the next 20K records (block comprising records 20K+1 to 40K) to the second thread, and the last 20K records (block comprising records 40K+1 to 60K) to the third thread. Assume the records need to be processed sequentially to avoid data corruption, while the processing logic in each thread is the same. Now, further assume a thread fails. A retry mechanism can be used and, to avoid clogging, a circuit breaker pattern can be implemented. However, the records of the failed thread will not get processed and no other service will automatically take the job. That is, conventional approaches rely on a manual intervention to attempt to resolve the issue and re-run the thread. However, such a conventional approach can adversely impact the data pipeline system.

This concept is generally illustrated in a computing environment100ofFIG.1. As shown, a source store (datastore)102is operatively coupled to an intermediate store104, which is operatively coupled to a final store106. In batch processing, parallel threads execute the workload as the data passes from source store102to intermediate store104(e.g., threads1,2, and3) and then to final store106(e.g., threads11,22, and33). In this non-limiting example, each thread processes a data subset (block) of a larger data set. That is, between source store102and intermediate store104, thread1processes a first subset of records (e.g., first 20K of 60K records), thread2processes a second subset of records (e.g., next 20K of 60K records), and thread3processes a third subset of records (e.g., last 20K of 60K records). Then, between intermediate store104and final store106, additional threads may process some subsets of the processed data stored at intermediate store104, e.g., thread11processes a first subset of records (e.g., first 10K records), thread22processes a second subset of records (e.g., next 10K records), and thread33processes a third subset of records (e.g., last 10K records). It is assumed that the logic in each of threads1,2, and3is the same, while the logic in each of threads11,22, and33is the same. Also, it is to be understood that the logic of threads1,2, and3can be the same as or different than the logic of threads11,22, and33.

In a container-based implementation, as generally depicted in a computing environment200ofFIG.2, each thread can be implemented as a microservice executing in a different container (i.e., the squares representing the containers). More particularly, as shown, a source store202is operatively coupled to an intermediate store204, which is operatively coupled to a final store206. Threads1,2,3,11,22, and33ofFIG.1are respectively depicted as microservices (MS)1,2,3,11,22, and33inFIG.2, with each MS being processed in its own container. Note that the microservices MS1, MS2, and MS3executed between source store202and intermediate store204are collectively referred to as Process1(203), while the microservices MS11, MS22, and MS33executed between intermediate store204and final store206are collectively referred to as Process2(205). It is assumed here that each container associated with MS1, MS2, and MS3has the same microservice deployed (only the name associated with the container is different), while each container associated with MS11, MS22, and MS33has the same microservice deployed (only the name associated with the container is different). As mentioned, MS1, MS2, MS3, MS11, MS22, and MS33can be different instances of the same microservice. Each container loads different record sets from one stage and transforms and loads them to the next stage.

As the term is illustratively used herein, a container may be considered lightweight, stand-alone, executable software code that includes elements needed to run the software code. The container structure has many advantages including, but not limited to, isolating the software code from its surroundings, and helping reduce conflicts between different tenants or users running different software code on the same underlying infrastructure. The term “user” herein is intended to be broadly construed so as to encompass numerous arrangements of human, hardware, software or firmware entities, as well as combinations of such entities.

In illustrative embodiments, containers may be implemented using a Kubernetes container orchestration system. Kubernetes is an open-source system for automating application deployment, scaling, and management within a container-based environment comprised of components referred to as pods, nodes and clusters, as will be further explained below. Types of containers that may be implemented or otherwise adapted within the Kubernetes system include, but are not limited to, Docker containers or other types of Linux containers (LXCs) or Windows containers. Kubernetes has become the prevalent container orchestration system for managing containerized workloads. It is rapidly being adopted by many enterprise-based information technology (IT) organizations to deploy its application programs (application). By way of example only, such applications may include both newly architected stateless or inherently redundant scale-out applications, as well as existing stateful applications. Non-limiting examples of stateful applications may include legacy databases such as Oracle, MySQL, and PostgreSQL, as well as other stateful applications that are not inherently redundant. While the Kubernetes container orchestration system is used to illustrate various embodiments, it is to be understood that alternative container orchestration systems can be utilized.

Some terminology associated with the Kubernetes container orchestration system will now be explained. In general, for a Kubernetes environment, one or more containers are part of a pod. Thus, the environment may be referred to, more generally, as a pod-based system, a pod-based container system, a pod-based container orchestration system, a pod-based container management system, or the like. As mentioned above, the containers can be any type of container, e.g., Docker container, etc. Furthermore, a pod is typically considered the smallest execution unit in the Kubernetes container orchestration environment. A pod encapsulates one or more containers. One or more pods are executed on a worker node. Multiple worker nodes form a cluster. A Kubernetes cluster is managed by a control plane node. By way of example, pods represent the respective processes running on a cluster.

In a Kubernetes container orchestration environment with autoscaling functionality, new pods are spun, and a system administrator cannot assign different sets of records (record sets or record blocks) to different pods individually, as there is no direct access to an individual pod. Thus, a system administrator deploys the same microservice in three different nodes and assigns the record set to each node, while disabling autoscaling functionality to avoid parallelism inside a given record set.

This concept is generally illustrated in a computing environment300ofFIG.3. As shown, a source store302is operatively coupled to an intermediate store304with MS1executing in a pod in node1, MS2executing in a pod of node2, and MS3executing in a pod of node3, collectively labeled as a cluster303.

Thus, as depicted in computing environment300, MS1, MS2and MS3are the same microservices deployed in different containers respectively executing on different pods of a cluster to mimic a controlled parallel process such that each container/pod processes different record sets from source store302.

Now assume that MS2fails. In response, a retry mechanism can be attempted. However, while the microservice is down, the records assigned to that microservice will not get processed and no other microservice will automatically take that work on. As a result, in a conventional approach, a system administrator needs to perform a manual intervention to fix the issue and re-run the microservice. This can have significant negative effects on the computing environment as including, but not limited to, time deadline misses, resource mismanagement, cost overruns, etc.

To further illustrate the technical issues, assume a computing environment400inFIG.4manages microservice1(402-1), a microservice2(402-2) and a microservice3(402-3) as different container deployments of the same logic (i.e., the same microservice deployed with a different name) and assigns the workload boundary linearly. In a conventional approach, the system administrator assigns a record set to each microservice which then, collectively, process the entire workload in a linear manner.

If the second container/thread fails (microservice2), the record sets assigned thereto (records of record set are circles depicted under microservice2) fail to process, and the other microservices will not automatically handle the unprocessed records assigned to microservice2. This leads to a manual intervention by a system administrator to fix microservice2and then to redeploy and restart the microservice to process the workload request (processing the assigned record set). In a critical workload use case, a system administrator cannot afford to take time to fix and re-run the failed microservice.

Illustrative embodiments address the above and other technical challenges in container-based microservice platforms such as, but not limited to, a Kubernetes platform. For example, illustrative data processing system embodiments provide improvements in batch job execution platforms by maintaining records to be processed in a ring architecture and assessing how many threads/containers are needed for processing. Then, the system distributes the different containers in the ring architecture to execute all records (given workload of data records) in a given direction around the ring, e.g., a clockwise direction. So, if one microservice container fails, the next microservice container in the ring architecture will automatically process the workload. Illustrative embodiments also enable a user to specify a predetermined time to complete a specific job using the fault-tolerant ring architecture.

More particularly, in contrast to the linear model of conventional workload management approaches (as described above), illustrative embodiments assess the number of microservice containers needed to process all records in the overall job, and provide a ring model for distributing workloads to be processed. Then, the microservice is distributed to multiple containers in a ring (circular) architecture for processing in the required/desired interval. For example, the microservice containers start processing records in a clockwise direction, until no further records for processing are found. As such, if a microservice container fails, another one of the microservice containers will process those records automatically.

This concept is illustrated inFIG.5as a ring model500for microservice computing according to an illustrative embodiment.FIGS.6A through6Ctabularly illustrate some of the steps associated with the ring model500ofFIG.5. The terms ring model and ring architecture can also be more generally referred to herein as a “ring configuration.”

As shown in ring model500, records are distributed in a ring-based manner with record set movement occurring in the same circular direction, e.g., clockwise. The system assesses the records to be executed keeping in mind the record sets that are to be executed in sequence and the total time allowed to execute all records. Assume the same example as described above of an entire data set comprised of 60K records where three record sets are identified and are to be executed in parallel. The three record sets are identified as: record 1 to record 20,000 (20K); record 20,001 (20K+1) to record 40,000 (40K); and record 40,001 (40K+1) to record 60,000 (60K). Thus, the entire data set (given workload of data records) to be processed is 60K records, with each data subset comprising 20K records.

In one illustrative embodiment, assume the system spins three threads, deploys three Docker containers of the same microservice image, and instantiates three pods in three different nodes of a Kubernetes cluster. The threads/containers are distributed in an appropriate node processing location of the assigned pods. Each microservice instance/container gets a next record to process as the records move clockwise in the ring architecture. In other words, each microservice/container executes the record set available in a clockwise direction until there is no longer a next record to process.

It is to be understood that, in illustrative embodiments, the system records the assignments and state of service as shown in table602ofFIG.6A.

The system then starts executing the data pipeline process. Assume, as shown in table602, that all service instances have a service status of active. When MS1reaches record 20K, it tries to get the next record. It may be that, at that time, MS1could conceivably get the 35Kth record if MS2, which is executing from 20K+1 to 40K, is currently executing 35K−1th record. Since the 35Kth record belongs to MS2and MS2is active, MS1stops executing as the system knows MS2is taking care of its own remaining records. It is to be understood that if there is no sequence processing issue within a record set (i.e., if record 35K does not have to be processed after record 35K−1 completes processing and before record 35K+1 starts processing), the system can allow MS1to execute 35K and onwards, in order to speed up the process by allowing both MS1and MS2to execute parts of the second block.

Assume now that MS2fails and therefore stops executing records (i.e., MS2is down). The system updates the service state of MS2as not active (inactive). This is shown in table604ofFIG.6B. In this case, MS1executes until record 20K is completed and, being configured to look for the next record, checks the service status table to see that MS2is inactive. MS1will then start executing the records of the record set that MS2was responsible for processing before going down, i.e., records from 20K+1 to 40K, until it does not get a next record. Thus, advantageously due to the system architecture implementing ring model500, MS1automatically processes the records to be executed by the inactive MS2.

Assume now that MS1fails and therefore stops executing records (i.e., MS1is down). The system updates the service state of MS1as not active (inactive), while MS2and MS3are active. This is shown in table606ofFIG.6C. In this case, MS3executes until record 60K is completed and, being configured to look for the next record, checks the service status table to see that MS1is inactive. MS3will then start executing the records of the record set that MS1was responsible for processing before going down, i.e., records from 1 to 20K, until it does not get a next record. Thus, advantageously due to the system architecture implementing ring model500, MS3automatically processes the records to be executed by the inactive MS1.

Although not expressly shown in the service status tables ofFIGS.6A-6C, if MS3becomes inactive, MS2would process the records assigned to MS3after completing the processing of its own records. In alternative scenarios, an active MS can start processing records of a record set assigned to an inactive MS at any point within the record set that the inactive MS went down.

Referring now toFIG.7, a ring architecture-based microservice computing system700(system700) according to an illustrative embodiment is shown. More particularly, system700comprises several modules configured to perform functions described herein including a job scheduler and workload distributer702, a service/container manager704, a service image registry706, a service/container logs store708, and a ring model710(configured the same as ring model500described above in the context ofFIG.5).

In one or more illustrative embodiments, job scheduler and workload distributer702is configured to schedule the workload to be run by system700(e.g., daily at 4 pm, or daily between 4 pm and 8 am, etc.). At some predetermined time before running the workload (e.g., ten minutes, etc.), job scheduler and workload distributer702assesses the number of records to be run in the workload and determines the number of blocks of records to be executed in parallel. Job scheduler and workload distributer702also notes (e.g., marks) whether a record set requires parallelism or not, as this can influence whether an MS in the ring model710can start at some point within another MS's record set or has to start at the beginning of the record set.

The information assessed/determined by job scheduler and workload distributer702is passed to service/container manager704. Service/container manager704is configured to, according to the number of record sets, create the microservice instances (in this example, MS1, MS2, and MS3) based on a standardized service image template from service image registry706. Each microservice is instantiated with the configuration parameter of “starting record set number,” e.g., MS1is configured to start at record 1, MS2is configured to start at record 20K+1, and MS3is configured to start at 40K+1. Service/container manager704also maintains the service status of each microservice/container instance it spins. That is, if any microservices is down, service/container manager704marks the microservice instance as not active and denotes the last record successfully executed by the microservice instance before going down. In one or more illustrative embodiments, service/container manager704maintains the service status and last record processed in service/container logs store708(e.g., data structures such as tables602,604, and606or other data structures can be updated by service/container manager704and stored in service/container logs store708).

Once one of the microservice instances (MS1, MS2, MS3) finishes its assigned job and seeks a next one, given that microservice instances get records to be executed in a clockwise direction in the ring model710, service/container manager704provides the information about the actual service status, i.e., whether it is inactive (e.g., paused, stopped, down, failed, offline, etc.) or active (e.g., up, on, processing, online, etc.).

It is to be appreciated that some workloads/jobs are required to be completed within a predetermined time range and/or by a given deadline. As such, in one or more illustrative embodiments, system700is further configured to assess and deploy the required/desired (e.g., best, optimal, appropriate, sufficient, etc.) number of microservice containers for processing a workload/job in a predetermined time.

By way of example, batch processing is typically executed as a job that comprises: read a record from a datastore or queue (source store) apply a data transformation (microservice) write the updated record in a target datastore (intermediate or final store).

Consider a data pipeline implementation in a supply chain management use case.FIGS.8A and8Billustrate a use case with which a ring architecture-based microservice computing system according to an illustrative embodiment can be implemented. More particularly, consider a computing environment800inFIG.8Awherein raw demand data802is obtained from a first datastore and pre-processed by a microservice container804, and the pre-processed data is stored as pre-processed demand data806in a second datastore. Then, pre-processed demand data806is obtained from the second datastore and processed by a microservice container808that implements a material planning function. The data processed by microservice container808is then stored as material planning data810in a third datastore.

Assume that microservice container808that executes the material planning function is selectively configured to start at 3 pm. Thus, before that time, microservice container804has to pre-process raw demand data802. Assuming pre-processing starts at 2:15 pm, pre-processing microservice container804should complete in no more than 44 minutes (less than 45 minutes because material planning execution begins at 3 pm).

In one or more illustrative embodiments, system700runs in a learn mode without a time bound. In the learn mode, in one illustrative embodiment, at least two containers are spun. In the learn mode, a system administrator is not enabled to enter a specific time for processing the total batch job.

When the job is executed, system700captures the time to process the records by each container. System700calculates the average time to process one record by a microservice container. Once the learned average time taken to process one record is computed, the learn mode ends. For example, assume system700learns the average time taken to process one raw demand data record by one microservice container is 250 seconds.

After the learn mode, system700can enter a run mode. Run mode, in one illustrative embodiment, can be entered by the system administrator changing job scheduler (e.g., part of702in system700) to this mode to enable entry of the time for processing the job. The run mode can also occur automatically after the learn mode in some embodiments. In the above example, the system administrator can set the pre-processing microservice job to 40 minutes (i.e., building in a safety buffer so that material planning execution can start at 3 pm).

When the next scheduled time comes, system700performs the following:(i) obtains the total number of records to be processed, e.g., assume 5000K records of raw demand data802are to be pre-processed);(ii) receive input from the system administrator the time to complete the job, e.g., 40 minutes;(iii) look up average time to process one record determined in the learn mode, e.g., 250 seconds; and(iv) compute the number of microservice containers required to complete pre-processing using the following formula:

Number of containers required=Total number of records/(Average time to process one record*60*User entered time).

Thus, in the above example, the computation is as follows:

Number of containers required=5000K/(250*60*40)=8.333 rounded up to nine containers.

Thus, service/container manager704of system700causes deployment of nine containers in the ring architecture as illustrated inFIG.8Bas ring model820. At runtime (in run mode), system700monitors the time to process a single record. If it is greater than the time computed in the learn mode (e.g., 250 seconds), system700recalculates the number of containers and adjusts the ring model (i.e., adding one or more microservice container instances). Advantageously, with the ring architecture, in run mode (at runtime), system700can insert a new microservice container at any point of time. Each microservice container will keep executing the records in a clockwise direction, until it has no further records to process.

FIG.9illustrates a ring architecture-based workload management methodology (methodology900) according to an illustrative embodiment. It is to be understood that, in illustrative embodiments, methodology900is performed by system700ofFIG.7.

As shown, step902determines a set of containers for processing a given workload of data records, wherein each container of the set of containers is configured to process a given subset of the data records. Step904causes deployment of the set of containers in a ring configuration to process the given workload of data records such that at least one of the containers in the ring configuration automatically processes one or more data records of a subset of data records assigned to another container in the ring configuration that becomes inactive. Step906then causes the processing of the given workload of data records in accordance with the set of containers deployed in the ring configuration.

The particular processing operations and other system functionality described in conjunction with the diagrams described herein are presented by way of illustrative example only, and should not be construed as limiting the scope of the disclosure in any way. Alternative embodiments can use other types of processing operations and messaging protocols. For example, the ordering of the steps may be varied in other embodiments, or certain steps may be performed at least in part concurrently with one another rather than serially. Also, one or more of the steps may be repeated periodically, or multiple instances of the methods can be performed in parallel with one another.

Illustrative embodiments of processing platforms utilized to implement functionality for ring architecture-based workload management in a microservice computing environment will now be described in greater detail with reference toFIGS.10and11. Although described in the context of systems/module/processes ofFIGS.1-9, these platforms may also be used to implement at least portions of other information processing systems in other embodiments.

FIG.10shows an example processing platform comprising cloud infrastructure1000. The cloud infrastructure1000comprises a combination of physical and virtual processing resources that may be utilized to implement at least a portion of system700. The cloud infrastructure1000comprises multiple container sets1002-1,1002-2, . . .1002-L implemented using virtualization infrastructure1004. The virtualization infrastructure1004runs on physical infrastructure1005, and illustratively comprises one or more hypervisors and/or operating system level virtualization infrastructure.

The cloud infrastructure1000further comprises sets of applications1010-1,1010-2, . . .1010-L running on respective ones of the container sets1002-1,1002-2, . . .1002-L under the control of the virtualization infrastructure1004. The container sets1002may comprise respective sets of one or more containers.

In some implementations of theFIG.10embodiment, the container sets1002comprise respective containers implemented using virtualization infrastructure1004that provides operating system level virtualization functionality, such as support for Kubernetes-managed containers.

As is apparent from the above, one or more of the processing modules or other components of system700may each run on a computer, server, storage device or other processing platform element. A given such element may be viewed as an example of what is more generally referred to herein as a “processing device.” The cloud infrastructure1000shown inFIG.10may represent at least a portion of one processing platform. Another example of such a processing platform is processing platform1100shown inFIG.11.

The processing platform1100in this embodiment comprises a portion of system700and includes a plurality of processing devices, denoted1102-1,1102-2,1102-3, . . .1102-K, which communicate with one another over a network1104.

The processing device1102-1in the processing platform1100comprises a processor1110coupled to a memory1112.

The memory1112may comprise random access memory (RAM), read-only memory (ROM), flash memory or other types of memory, in any combination. The memory1112and other memories disclosed herein should be viewed as illustrative examples of what are more generally referred to as “processor-readable storage media” storing executable program code of one or more software programs.

Also included in the processing device1102-1is network interface circuitry1114, which is used to interface the processing device with the network1104and other system components, and may comprise conventional transceivers.

The other processing devices1102of the processing platform1100are assumed to be configured in a manner similar to that shown for processing device1102-1in the figure.

Again, the particular processing platform1100shown in the figure is presented by way of example only, and systems/modules/processes ofFIGS.1-9may include additional or alternative processing platforms, as well as numerous distinct processing platforms in any combination, with each such platform comprising one or more computers, servers, storage devices or other processing devices.

In some embodiments, storage systems may comprise at least one storage array implemented as a Unity, PowerMax, PowerFlex (previously ScaleIO) or PowerStore storage array, commercially available from Dell Technologies. As another example, storage arrays may comprise respective clustered storage systems, each including a plurality of storage nodes interconnected by one or more networks. An example of a clustered storage system of this type is an XtremIO™ storage array from Dell Technologies, illustratively implemented in the form of a scale-out all-flash content addressable storage array.