Dynamic scaling of computing message architecture

A method of processing data message in a computer system comprising a plurality of message queuing nodes, message processing nodes, and coordinating nodes. The method comprises adding a node to the computing system, determining the IP address and node identity number of the added node by a monitor Java archive (JAR) executing on a coordinating node of the computing system, sending the IP address and node identity number via a Java management extension (JMX) connector client of the monitor JAR to each of the message queuing nodes and message processing nodes via a Java management extension (JMX) connector associated with an MBean of each of the message queuing nodes and message processing nodes, where the MBeans encapsulates a configuration file, whereby each of the configuration files are updated with the IP address and the node identity number of the added message queuing node.

CROSS-REFERENCE TO RELATED APPLICATIONS

Not applicable.

REFERENCE TO A MICROFICHE APPENDIX

Not applicable.

BACKGROUND

Computer systems may generate large flows of data. Different computers within the computer system may be used to implement architectural layers of a composite data flow, where the data may flow from a data source to a first layer, from the first layer to a second layer, from the second layer to other layers and ultimately to a final layer. Some of the layers may provide a message queuing function to decouple a lower layer from a higher layer, so the two layers need not synchronize to coordinate the flow of data. For example, a lower layer may push a message to the message queue, where the message comprises some data or information to be processed by a higher layer. The higher layer retrieves the message from the message queue when it is ready to process it. If the message queue is empty, the higher layer waits until another message is available. In this way the lower layer does not have to handshake with the higher layer to pass messages.

SUMMARY

In an embodiment, a method of processing data messages in a computing system is disclosed. The method comprises receiving messages by a plurality of message queuing nodes of the computing system and storing the messages in message queues provided by the message queuing nodes, where each of the message queuing nodes execute a message queuing service and each message queuing service comprises a Java management bean (MBean). MBean encapsulates a message queuing service configuration file that identifies other nodes in the computing system by IP address and node identity number. The method further comprises retrieving messages from the message queues by a plurality of message processing nodes of the computing system. The method further comprises processing the retrieved messages, and delivering these processed messages, where each of the message processing nodes execute a message processing service and each message processing service comprises an MBean that encapsulates a message processing service configuration file that identifies other nodes in the computing system by IP address and node identity number, and adding a message queuing node to the computing system. The method further comprises detecting the addition of the added message queuing node by a monitor Java archive (JAR) executing on a coordinating node of the computing system and determining an IP address and a node identity number of the added message queuing node by the monitor JAR. The method further comprises sending the IP address and the node identity number of the added message queuing node via a Java management extension (JMX) connector client of the monitor JAR to each of the message queuing nodes via a JMX connector associated with the MBean of the message queuing service executing on that node, whereby the message queuing service configuration files of the message queuing services are updated with the IP address and the node identity number of the added message queuing node. The method further comprises sending the IP address and the node identity number of the added message queuing node via a JMX connector client of the monitor JAR to each of the message processing nodes via a JMX connector associated with the MBean of the message processing service executing on that node. By this process, the message processing service configuration files of the message processing services are updated with the IP address and the node identity number of the added message queuing node, and a message queuing layer of the computing system is scaled up.

In another embodiment, a method of processing data messages in a computing system is disclosed. The method comprises receiving messages by a plurality of message queuing nodes of the computing system and storing the messages in message queues provided by the message queuing nodes, where each of the message queuing nodes execute a message queuing service and each message queuing service comprises a Java management bean (MBean). The MBean encapsulates a message queuing service configuration file that identifies other nodes in the computing system by IP address and node identity number. The method further comprises retrieving messages from the message queues by a plurality of message processing nodes of the computing system, processing the retrieved messages, and delivering these processed messages. Each of the message processing nodes execute a message processing service, and each message processing service comprises an MBean that encapsulates a message processing service configuration file that identifies other nodes in the computing system by IP address and node identity number. The method further comprises adding a message processing node to the computing system. The method further comprises detecting the addition of the added message processing node by a monitor Java archive (JAR) executing on a coordinating node of the computing system, determining an IP address and a node identity number of the added message processing node by the monitor JAR, sending the IP address and the node identity number of the added message processing node via a Java management extension (JMX) connector client of the monitor JAR to each of the message queuing nodes via a JMX connector associated with the MBean of the message queuing service executing on that node. The message queuing service configuration files of the message queuing services thereby are updated with the IP address and the node identity number of the added message processing node. The method further comprises sending the IP address and the node identity number of the added message processing node via a JMX connector client of the monitor JAR to each of the message processing nodes via a JMX connector associated with the MBean of the message processing service executing on that node, whereby the message processing service configuration files of the message processing services are updated with the IP address and the node identity number of the added message processing node. Through this process, a message processing layer of the computing system is scaled up.

In yet another embodiment, a method of processing data messages in a computing system is disclosed. The method comprises receiving messages by a plurality of message queuing nodes of the computing system and storing the messages in message queues provided by the message queuing nodes, where each of the message queuing nodes execute a message queuing service and each message queuing service comprises a Java management bean (MBean). The MBean encapsulates a message queuing service configuration file that identifies other nodes in the computing system by IP address and node identity number, retrieving messages from the message queues by a plurality of message processing nodes of the computing system. The method further comprises processing the retrieved messages. The method further comprises delivering these processed messages, where each of the message processing nodes execute a message processing service and each message processing service comprises an MBean that encapsulates a message processing service configuration file that identifies other nodes in the computing system by IP address and node identity number. The method further comprises removing a message queuing node or a message processing node from the computing system. The method further comprises detecting the removal of the message queuing node by a monitor Java archive (JAR) executing on a coordinating node of the computing system and determining an IP address and a node identity number of the removed message queuing node by the monitor JAR. The method further comprises sending the IP address and the node identity number of the removed message queuing node via a Java management extension (JMX) connector client of the monitor JAR to each of the remaining message queuing nodes via a JMX connector associated with the MBean of the message queuing service executing on that node, whereby the message queuing service configuration files of the message queuing services are updated by removing the IP address and the node identity number of the removed message queuing node. The method further comprises sending the IP address and the node identity number of the removed message queuing node via a JMX connector client of the monitor JAR to each of the message processing nodes via a JMX connector associated with the MBean of the message processing service executing on that node, whereby the message processing service configuration files of the message processing services are updated by removing the IP address and the node identity number of the removed message queuing node. By this process a message queuing layer of the computing system is scaled down.

DETAILED DESCRIPTION

The present disclosure teaches a tiered messaging system that automatically adapts configuration files and causes services to take cognizance of the reconfigured computing nodes as computing nodes are added to the messaging system or removed from the messaging system. The system may receive a heavy stream of data messages from a plurality of producer processes not formally considered part of the system. A message queuing layer receives and queues these messages. A message processing layer, mediated by a coordinating layer, retrieves messages from the message queuing layer and processes these messages before storing them in one or more data stores which are not formally considered part of the system. The producers may be forwarding event messages associated with a large number of mobile communication devices, for example tens of millions of mobile communication devices. The processed messages may be processed by dropping some messages, aggregating other messages, and creating reports that summarize a plurality of other messages. The message queuing layer, the coordinating layer, and the message processing layer each comprise a plurality of computing nodes, and the number of these computing nodes may be increased or decreased as message streaming loads increase or decrease. Computing nodes in the message queuing layer are referred to as message queue nodes, computing nodes in the message processing layer are referred to as message processing nodes, and computing nodes in the coordinating layer are referred to as coordinating nodes. In an embodiment, the computing nodes are provided by a cloud computing system, and the message queuing, coordinating, and message processing functionality is provided by services running on virtual machines provided on the computing nodes.

When a computing node is added or removed from the tiered messaging system, configuration files associated with each of the other computing nodes in the tiered messaging system desirably are updated with the change and the associated service supported by that computing node is caused to take note of the change and adapt its activity accordingly. For example, an IP address and node identity number of the added computing node is added to the configuration files or the IP address and node identity number of the removed computing node is removed from the configuration files and then the services on the remaining computing nodes are adapted to the updated configuration files. For example, lists of messaging queues from which to retrieve messages maintained by message processing services, for example data structures relied upon by the message processing services, may be modified by a dedicated routine or method of the message processing service. For example, lists of message processing queues known to the message queuing service may be modified by routines or methods of the message queuing service. For example, message topics may be redistributed among messaging queues to adapt to a changed number of messaging queues as reflected in the associated configuration files. These adaptations can take place automatically when the configuration files have been completed and each respective service executes a service-specific routine for dynamically adapting to the changed configuration files.

In the past this modification of computing nodes was handled manually by human system administrators or technicians and may have entailed restarting the services. In an example system, there may be about 25 computing nodes running in the message queuing layer, 5 computing nodes running in the mediation layer, and 20 computing nodes running in the message processing layer. The number of computing nodes may diurnally cycle through an increase of computing capacity by adding nodes and a decrease of computing capacity by removing nodes, as the activity of mobile communication devices diurnally cycles through a peak of activity and a minimum of activity. This manual process consumed the time of technicians undesirably and further impeded the responsiveness of the layered messaging system. The present disclosure teaches automatically adapting the configuration files and each service automatically adapting to the changed configuration files without restarting and without the continued involvement of the technician other than to initially trigger the update, for example by adding a new computing node or by removing an existing computing node and notifying a coordinating agent executing in the coordination layer.

A Java archive (JAR) executes on one of a plurality of the coordinating nodes. This JAR may be referred to as a monitor JAR in some contexts herein. The monitor JAR monitors the computing system and the configuration files of each of the computing nodes. When a new node is added to the system or when an existing node is removed from the system, the monitor JAR detects this event and causes the rest of the system to adapt accordingly. When the monitor JAR detects a new node it invokes an application programming interface (API) of a Java management extension (JMX) management bean (MBean) in each of the computing nodes to update a configuration file that is encapsulated in the MBean of each computing node. More specifically, the monitor JAR comprises a JMX connector client that communicates with a JMX connector in a computing node. In an embodiment, the MBeans may register themselves with the JAR as their computing node is added to the system and brought into service.

The monitor JAR sends a message via the JMX connector client to the JMX connector in the computing node where the message identifies the IP address and the node identity number of the added computing node. The JMX connector in the computing node passes this information to the MBean, for example via the intermediary of MBean Server of the computing node. This process causes the configuration file encapsulated by the MBean to be updated. This also entails the service executing on the subject computing node to take cognizance of the revised configuration file and adapt its execution accordingly. The monitor JAR may have a single JMX connector client that communicates with a plurality of different JMX connectors on different computing nodes. Alternatively, the monitor JAR may have a different JMX connector client for each different JMX connector on the different computing nodes. When a computing node is removed from the system (i.e., a computing node is turned off or removed from service) the process works in the same way but the communication from the JMX connector client informs the JMX connector on each computing node that the IP address and the node identity number in the message ought to be removed from the configuration file.

This system promotes system resources tracking current processing demands more faithfully and efficiently. For example, when a human administrator scales the system up or down, they must take into account the time it takes them to step through the process manually, and this implies a time lagging function that does not comport with scaling down as far as might be desirable. Thus, the human user typically over provisions resources, because they fear they cannot scale up timely when demand increases.

The monitor JAR determines additions or deletions of nodes promptly. It promptly determines what queuing nodes it should update with the changes and updates them, serially and promptly. It then promptly determines what processing nodes it should update with the changes and updates them, serially and promptly. This solution does not entail any downtime or maintenance time that interrupts service to a user community.

Turning now toFIG. 1, a system100is described. In an embodiment, system100comprises a plurality of message producers102, a network104, a plurality of message queuing nodes106, a plurality of coordinating nodes114, and a plurality of message processing nodes122. The network104comprises one or more public communication networks, one or more private communication networks, or a combination thereof. Each message queuing node106comprises a message queuing service108, a queuing Java management extensions (JMX) connector110, and a message queuing node configuration MBean112that manages a message queuing node configuration file113. Each coordinating node114comprises a coordinating service116, a coordinating JMX connector118, and a coordinating node configuration MBean120that manages a coordinating node configuration file121. In an embodiment, one of the coordinating nodes114further comprises a monitor Java archive (JAR)123that comprises at least one JMX connector client125. Each message processing node122comprises a message processing service124, a processing JMX connector126, and a message processing node configuration MBean128that manages a message processing node configuration file121. The MBeans112,120,128may be said to encapsulate the respective configuration files113,121,129.

Each node106,114,122may be referred to as a computing node and is a computer system. Computer systems are described hereinafter. Each of the services108,116,124are provided by one or more applications or computer programs executing on a virtual machine or a plurality of virtual machines on its subject node.

Each of the services108,116,124learns of other services108,116,124and nodes106,114,122and collaborates with them based on a configuration file113,121,129encapsulated in an MBean112,120,128local to the node106,114,122on which it executes. When the configuration file113,121,129on a node106,114,122is changed, the service108,116,124executing on the same node106,114,122may be adapted based on the now changed configuration file113,121,129, which in turn may alter its collaboration with other nodes106,114,122in the system102. For example, a routine of the services may read the changed configuration file and adapt behavior of the service accordingly. In an embodiment, the message queuing service108is implemented as a Kafka message queuing service or system. In an embodiment, the message processing service124is implemented using Storm. In an embodiment, the coordinating service116is implemented using Zookeeper.

The message producers102may be applications that execute on computer systems. The message producers102send messages relating events which have occurred on or related to mobile communication devices via the network104to the message queuing service108on the message queuing nodes106for storage while waiting for processing by message processing service124on the message processing nodes122. The message producers102may receive events from tens of millions of mobile communication devices. The events may comprise signal quality data, cell site signal strength data, requests for a voice service link, requests for a data service link, a URL sent in an HTTP request, a position of the mobile communication device, a blocked call attempt, a dropped call, a hand-off record, and other events. A single mobile communication device may generate 100s or even 1000s of events per day, and these events from tens of millions of devices may be streamed by the message producers102into the message queuing service108.

The message processing service124fetches messages enqueued by the message queuing service108, with the help of the coordinating service116, and does some processing on the mobile communication device event encapsulated in the fetched messages. This processing may include counting the events by category (e.g., number of dropped calls, number of blocked call attempts, number of successful calls), aggregating separate events into a single event (e.g., multiple events related to a single call combined in a single call record), storing events in storage or in a data store130, placing the events in long term archival storage.

As the influx of messages from the message producers102increases or decreases, the number of computing nodes in the system100increases or decreases. When a node106,114,122is added or removed from the system100, the on-going services108,116,124learn of the change through the changes in their respective configuration files113,121,129and adapt to the change accordingly. In an embodiment, this is accomplished by first modifying the configuration file110,118,126and then the services108,116,124execute internal routines that take cognizance of the changed configuration files110,118,126. Formerly the modification of configuration files110,118,126and restarting of services108,116,124was handled manually. The present disclosure teaches a specific information technology solution for performing this activity automatically, whereby to reduce the burden on IT staffs, to reduce the opportunities for error, and to make the system100scale more rapidly.

When a new computing node is brought up and introduced into the system, the monitor JAR123detects the new node and learns its IP address and node identity number. The monitor JAR123then updates the configuration files113,121,129of the message queuing nodes106, the coordinating nodes114, and the message processing nodes122. In each case, this updating is accomplished by the monitor JAR123causing the JMX connector client125to communicate with a JMX connector110,118,126in the node106,114,122providing the IP address and node identity number. The JMX connector110,118,126invokes an API of the configuration MBean112,120,128to cause the IP address and the node identity number of the added node to be added to the encapsulated configuration file113,121,129. The service108,116,124may detect the update to the encapsulated configuration file113,121,129and execute a routine that adapts the service108,116,124so as to continue processing going forwards based on the changed configuration files113,121,129. This process works in much the same way when a node106,114,122is removed from the system100(e.g., is turned off or taken out of service), except that in this case the IP address and node identity number are removed from the configuration file113,131,129rather than added.

Turning now toFIG. 2AandFIG. 2B, a method200is described. At block202, a plurality of message queuing nodes of the computing system receive and store messages in message queues provided by the message queuing nodes, where each of the message queuing nodes execute a message queuing service and each message queuing service comprises a Java management bean (MBean) that encapsulates a message queuing service configuration file that identifies other nodes in the computing system by IP address and node identity number. At block204, a plurality of message processing nodes of the computing system retrieve messages from the message queues, processing the retrieved messages, and delivering these processed messages, where each of the message processing nodes execute a message processing service and each message processing service comprises an MBean that encapsulates a message processing service configuration file that identifies other nodes in the computing system by IP address and node identity number.

At block206, add a message queuing node to the computing system. At block208, a monitor Java archive (JAR) executing on a coordinating node of the computing system detects the addition of the added message queuing node. At block210, the monitor JAR determines an IP address and a node identity number of the added message queuing node.

At block212, a Java management extension (JMX) connector client of the monitor JAR sends the IP address and the node identity number of the added message queuing node to each of the message queuing nodes via a JMX connector associated with the MBean of the message queuing service executing on that node, whereby the message queuing service configuration files of the message queuing services are updated with the IP address and the node identity number of the added message queuing node. At block214, a JMX connector client of the monitor JAR sends the IP address and the node identity number of the added message queuing node to each of the message processing nodes via a JMX connector associated with the MBean of the message processing service executing on that node.

Turning now toFIG. 3AandFIG. 3B, a method220is described. At block222, a plurality of message queuing nodes of the computing system receive and store messages in message queues provided by the message queuing nodes, where each of the message queuing nodes execute a message queuing service and each message queuing service comprises a Java management bean (MBean) that encapsulates a message queuing service configuration file that identifies other nodes in the computing system by IP address and node identity number. At block224, a plurality of message processing nodes of the computing system retrieve messages from the message queues, processing the retrieved messages, and delivering these processed messages, where each of the message processing nodes execute a message processing service and each message processing service comprises an MBean that encapsulates a message processing service configuration file that identifies other nodes in the computing system by IP address and node identity number.

At block226, add a message processing node to the computing system. At block228, a monitor Java archive (JAR) executing on a coordinating node of the computing system detects the addition of the added message processing node. At block230, the monitor JAR determines an IP address and a node identity number of the added message processing node

At block232, a Java management extension (JMX) connector client of the monitor JAR sends the IP address and the node identity number of the added message processing node to each of the message queuing nodes via a JMX connector associated with the MBean of the message queuing service executing on that node, whereby the message queuing service configuration files of the message queuing services are updated with the IP address and the node identity number of the added message processing node. At block234, a JMX connector client of the monitor JAR sends the IP address and the node identity number of the added message processing node to each of the message processing nodes via a JMX connector associated with the M-bean of the message processing service executing on that node, whereby the message processing service configuration files of the message processing services are updated with the IP address and the node identity number of the added message processing node.

Turning now toFIG. 4AandFIG. 4B, a method240is described. At block242, a plurality of message queuing nodes of the computing system receive and store messages in message queues provided by the message queuing nodes, where each of the message queuing nodes execute a message queuing service and each message queuing service comprises a Java management bean (MBean) that encapsulates a message queuing service configuration file that identifies other nodes in the computing system by IP address and node identity number. At block244, a plurality of message processing nodes of the computing system retrieve messages from the message queues, processing the retrieved messages, and delivering these processed messages, where each of the message processing nodes execute a message processing service and each message processing service comprises an MBean that encapsulates a message processing service configuration file that identifies other nodes in the computing system by IP address and node identity number. At block246, remove a message queuing node from the computing system.

At block248, a monitor Java archive (JAR) executing on a coordinating node of the computing system detects the removal of the added message processing node. At block250, the monitor JAR determines an IP address and a node identity number of the removed message queuing node.

At block252, a Java management extension (JMX) connector client of the monitor JAR sends the IP address and the node identity number of the removed message queuing node to each of the message queuing nodes via a JMX connector associated with the MBean of the message queuing service executing on that node, whereby the message queuing service configuration files of the message queuing services are updated by removing the IP address and the node identity number of the removed message queuing node. At block254, a JMX connector client of the monitor JAR sends the IP address and the node identity number of the removed message queuing node to each of the message processing nodes via a JMX connector associated with the M-bean of the message processing service executing on that node, whereby the message processing service configuration files of the message processing services are updated by removing the IP address and the node identity number of the removed message queuing node.