[Notes] [Git][BuildGrid/buildgrid][mablanch/135-monitoring-bus] 3 commits: monitoring.py: New monitoring bus class

[Martin Blanchard <gitlab mg gitlab com>]

Title: GitLab

Martin Blanchard pushed to branch mablanch/135-monitoring-bus at BuildGrid / buildgrid

Commits:

• 6537e098
  by Martin Blanchard at 2018-11-14T16:52:59Z

  monitoring.py: New monitoring bus class
  

• 26e005f8
  by Martin Blanchard at 2018-11-14T16:53:01Z

  Handle the server's main event loop internally
  

• af1414df
  by Martin Blanchard at 2018-11-14T16:53:01Z

  server/instance.py: Run monitoring bus in main event loop
  

3 changed files:

• buildgrid/_app/commands/cmd_server.py
• + buildgrid/server/_monitoring.py
• buildgrid/server/instance.py

Changes:

buildgrid/_app/commands/cmd_server.py

---

...	...	@@ -20,7 +20,6 @@ Server command
20	20	Create a BuildGrid server.
21	21	"""
22	22
23		-import asyncio
24	23	import logging
25	24	import sys
26	25
...	...	@@ -52,18 +51,14 @@ def start(context, config):
52	51	click.echo("ERROR: Could not parse config: {}.\n".format(str(e)), err=True)
53	52	sys.exit(-1)
54	53
55		- loop = asyncio.get_event_loop()
56	54	try:
57	55	server.start()
58		- loop.run_forever()
59	56
60	57	except KeyboardInterrupt:
61	58	pass
62	59
63	60	finally:
64		- context.logger.info("Stopping server")
65	61	server.stop()
66		- loop.close()
67	62
68	63
69	64	def _create_server_from_config(config):

buildgrid/server/_monitoring.py

---

	1	+# Copyright (C) 2018 Bloomberg LP
	2	+#
	3	+# Licensed under the Apache License, Version 2.0 (the "License");
	4	+# you may not use this file except in compliance with the License.
	5	+# You may obtain a copy of the License at
	6	+#
	7	+# <http://www.apache.org/licenses/LICENSE-2.0>
	8	+#
	9	+# Unless required by applicable law or agreed to in writing, software
	10	+# distributed under the License is distributed on an "AS IS" BASIS,
	11	+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	12	+# See the License for the specific language governing permissions and
	13	+# limitations under the License.
	14	+
	15	+
	16	+import asyncio
	17	+
	18	+from google.protobuf import json_format
	19	+
	20	+from buildgrid._protos.buildgrid.v2 import monitoring_pb2
	21	+
	22	+
	23	+class MonitoringBus:
	24	+
	25	+ def __init__(self, loop):
	26	+ self.__event_loop = loop
	27	+ self.__streaming_task = None
	28	+
	29	+ self.__message_queue = asyncio.Queue(loop=loop)
	30	+ self.__sequence_number = 1
	31	+
	32	+ # --- Public API ---
	33	+
	34	+ def start(self):
	35	+ """Starts the monitoring bus worker task."""
	36	+ self.__streaming_task = asyncio.ensure_future(
	37	+ self._streaming_worker(), loop=self.__event_loop)
	38	+
	39	+ def stop(self):
	40	+ """Cancels the monitoring bus worker task."""
	41	+ if self.__streaming_task is not None:
	42	+ self.__streaming_task.cancel()
	43	+
	44	+ async def send_record(self, record):
	45	+ """Publishes a record onto the bus asynchronously.
	46	+
	47	+ Args:
	48	+ record (Message): The
	49	+ """
	50	+ await self.__message_queue.put(record)
	51	+
	52	+ def send_record_nowait(self, record):
	53	+ """Publishes a record onto the bus.
	54	+
	55	+ Args:
	56	+ record (Message): The
	57	+ """
	58	+ self.__message_queue.put_nowait(record)
	59	+
	60	+ # --- Private API ---
	61	+
	62	+ async def _streaming_worker(self):
	63	+ """Handles bus messages steaming work."""
	64	+ async def __streaming_worker():
	65	+ record = await self.__message_queue.get()
	66	+
	67	+ message = monitoring_pb2.BusMessage()
	68	+ message.sequence_number = self.__sequence_number
	69	+
	70	+ if record.DESCRIPTOR is monitoring_pb2.LogRecord.DESCRIPTOR:
	71	+ message.log_record.CopyFrom(record)
	72	+
	73	+ elif record.DESCRIPTOR is monitoring_pb2.MetricRecord.DESCRIPTOR:
	74	+ message.metric_record.CopyFrom(record)
	75	+
	76	+ else:
	77	+ return False
	78	+
	79	+ # print(json_format.MessageToJson(message))
	80	+
	81	+ return True
	82	+
	83	+ try:
	84	+ while True:
	85	+ if await __streaming_worker():
	86	+ self.__sequence_number += 1
	87	+
	88	+ except asyncio.CancelledError:
	89	+ pass

buildgrid/server/instance.py

---

...	...	@@ -13,18 +13,21 @@
13	13	# limitations under the License.
14	14
15	15
	16	+import asyncio
16	17	from concurrent import futures
17	18	import logging
18	19	import os
	20	+import time
19	21
20	22	import grpc
21	23
22		-from .cas.service import ByteStreamService, ContentAddressableStorageService
23		-from .actioncache.service import ActionCacheService
24		-from .execution.service import ExecutionService
25		-from .operations.service import OperationsService
26		-from .bots.service import BotsService
27		-from .referencestorage.service import ReferenceStorageService
	24	+from buildgrid.server.actioncache.service import ActionCacheService
	25	+from buildgrid.server.bots.service import BotsService
	26	+from buildgrid.server.cas.service import ByteStreamService, ContentAddressableStorageService
	27	+from buildgrid.server.execution.service import ExecutionService
	28	+from buildgrid.server._monitoring import MonitoringBus
	29	+from buildgrid.server.operations.service import OperationsService
	30	+from buildgrid.server.referencestorage.service import ReferenceStorageService
28	31
29	32
30	33	class BuildGridServer:
...	...	@@ -34,7 +37,7 @@ class BuildGridServer:
34	37	requisite services.
35	38	"""
36	39
37		- def __init__(self, max_workers=None):
	40	+ def __init__(self, max_workers=None, monitor=True):
38	41	"""Initializes a new :class:`BuildGridServer` instance.
39	42
40	43	Args:
...	...	@@ -46,9 +49,11 @@ class BuildGridServer:
46	49	# Use max_workers default from Python 3.5+
47	50	max_workers = (os.cpu_count() or 1) * 5
48	51
49		- server = grpc.server(futures.ThreadPoolExecutor(max_workers))
	52	+ self.__grpc_executor = futures.ThreadPoolExecutor(max_workers)
	53	+ self.__grpc_server = grpc.server(self.__grpc_executor)
50	54
51		- self._server = server
	55	+ self.__main_loop = asyncio.get_event_loop()
	56	+ self.__monitoring_bus = None
52	57
53	58	self._execution_service = None
54	59	self._bots_service = None
...	...	@@ -58,15 +63,35 @@ class BuildGridServer:
58	63	self._cas_service = None
59	64	self._bytestream_service = None
60	65
	66	+ self._is_monitored = monitor
	67	+
	68	+ if self._is_monitored:
	69	+ self.__monitoring_bus = MonitoringBus(self.__main_loop)
	70	+
	71	+ # --- Public API ---
	72	+
61	73	def start(self):
62		- """Starts the server.
	74	+ """Starts the BuildGrid server.
63	75	"""
64		- self._server.start()
	76	+ self.__grpc_server.start()
	77	+
	78	+ if self._is_monitored:
	79	+ self.__monitoring_bus.start()
	80	+ self.__main_loop.run_forever()
65	81
66	82	def stop(self, grace=0):
67		- """Stops the server.
	83	+ """Stops the BuildGrid server.
	84	+
	85	+ Args:
	86	+ grace (int, optional): A duration of time in seconds. Defaults to 0.
68	87	"""
69		- self._server.stop(grace)
	88	+ if self._is_monitored:
	89	+ self.__monitoring_bus.stop()
	90	+
	91	+ self.__grpc_server.stop(grace)
	92	+
	93	+ if grace > 0:
	94	+ time.sleep(grace)
70	95
71	96	def add_port(self, address, credentials):
72	97	"""Adds a port to the server.
...	...	@@ -80,11 +105,11 @@ class BuildGridServer:
80	105	"""
81	106	if credentials is not None:
82	107	self.__logger.info("Adding secure connection on: [%s]", address)
83		- self._server.add_secure_port(address, credentials)
	108	+ self.__grpc_server.add_secure_port(address, credentials)
84	109
85	110	else:
86	111	self.__logger.info("Adding insecure connection on [%s]", address)
87		- self._server.add_insecure_port(address)
	112	+ self.__grpc_server.add_insecure_port(address)
88	113
89	114	def add_execution_instance(self, instance, instance_name):
90	115	"""Adds an :obj:`ExecutionInstance` to the service.
...	...	@@ -96,7 +121,7 @@ class BuildGridServer:
96	121	instance_name (str): Instance name.
97	122	"""
98	123	if self._execution_service is None:
99		- self._execution_service = ExecutionService(self._server)
	124	+ self._execution_service = ExecutionService(self.__grpc_server)
100	125
101	126	self._execution_service.add_instance(instance_name, instance)
102	127
...	...	@@ -110,7 +135,7 @@ class BuildGridServer:
110	135	instance_name (str): Instance name.
111	136	"""
112	137	if self._bots_service is None:
113		- self._bots_service = BotsService(self._server)
	138	+ self._bots_service = BotsService(self.__grpc_server)
114	139
115	140	self._bots_service.add_instance(instance_name, instance)
116	141
...	...	@@ -124,7 +149,7 @@ class BuildGridServer:
124	149	instance_name (str): Instance name.
125	150	"""
126	151	if self._operations_service is None:
127		- self._operations_service = OperationsService(self._server)
	152	+ self._operations_service = OperationsService(self.__grpc_server)
128	153
129	154	self._operations_service.add_instance(instance_name, instance)
130	155
...	...	@@ -138,7 +163,7 @@ class BuildGridServer:
138	163	instance_name (str): Instance name.
139	164	"""
140	165	if self._reference_storage_service is None:
141		- self._reference_storage_service = ReferenceStorageService(self._server)
	166	+ self._reference_storage_service = ReferenceStorageService(self.__grpc_server)
142	167
143	168	self._reference_storage_service.add_instance(instance_name, instance)
144	169
...	...	@@ -152,7 +177,7 @@ class BuildGridServer:
152	177	instance_name (str): Instance name.
153	178	"""
154	179	if self._action_cache_service is None:
155		- self._action_cache_service = ActionCacheService(self._server)
	180	+ self._action_cache_service = ActionCacheService(self.__grpc_server)
156	181
157	182	self._action_cache_service.add_instance(instance_name, instance)
158	183
...	...	@@ -166,7 +191,7 @@ class BuildGridServer:
166	191	instance_name (str): Instance name.
167	192	"""
168	193	if self._cas_service is None:
169		- self._cas_service = ContentAddressableStorageService(self._server)
	194	+ self._cas_service = ContentAddressableStorageService(self.__grpc_server)
170	195
171	196	self._cas_service.add_instance(instance_name, instance)
172	197
...	...	@@ -180,6 +205,12 @@ class BuildGridServer:
180	205	instance_name (str): Instance name.
181	206	"""
182	207	if self._bytestream_service is None:
183		- self._bytestream_service = ByteStreamService(self._server)
	208	+ self._bytestream_service = ByteStreamService(self.__grpc_server)
184	209
185	210	self._bytestream_service.add_instance(instance_name, instance)
	211	+
	212	+ # --- Public API: Monitoring ---
	213	+
	214	+ @property
	215	+ def is_monitored(self):
	216	+ return self._is_monitored

—
View it on GitLab.
You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.