Coverage for /builds/BuildGrid/buildgrid/buildgrid/server/operations/instance.py: 90.38%

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""" 

17OperationsInstance 

18================== 

19An instance of the LongRunningOperations Service. 

20""" 

21 

22import logging 

23 

24from buildgrid._exceptions import InvalidArgumentError, NotFoundError 

25from buildgrid._protos.google.longrunning import operations_pb2 

26from buildgrid.server.operations.filtering import FilterParser, DEFAULT_OPERATION_FILTERS 

27from buildgrid.server.metrics_names import ( 

28 OPERATIONS_CANCEL_OPERATION_TIME_METRIC_NAME, 

29 OPERATIONS_GET_OPERATION_TIME_METRIC_NAME, 

30 OPERATIONS_LIST_OPERATIONS_TIME_METRIC_NAME 

31) 

32from buildgrid.server.metrics_utils import DurationMetric 

33from buildgrid.settings import DEFAULT_MAX_LIST_OPERATION_PAGE_SIZE 

34 

35 

36class OperationsInstance: 

37 

38 def __init__(self, scheduler, max_list_operations_page_size=DEFAULT_MAX_LIST_OPERATION_PAGE_SIZE): 

39 self.__logger = logging.getLogger(__name__) 

40 

41 self._scheduler = scheduler 

42 self._max_list_operations_page_size = max_list_operations_page_size 

43 self._instance_name = None 

44 

45 # --- Public API --- 

46 

47 @property 

48 def instance_name(self): 

49 return self._instance_name 

50 

51 @property 

52 def scheduler(self): 

53 return self._scheduler 

54 

55 def setup_grpc(self): 

56 # The operations instance doesn't currently have any gRPC setup to do 

57 pass 

58 

59 def register_instance_with_server(self, instance_name, server): 

60 """Names and registers the operations instance with a given server.""" 

61 if self._instance_name is None: 

62 server.add_operations_instance(self, instance_name) 

63 

64 self._instance_name = instance_name 

65 

66 else: 

67 raise AssertionError("Instance already registered") 

68 

69 @DurationMetric(OPERATIONS_GET_OPERATION_TIME_METRIC_NAME, instanced=True) 

70 def get_operation(self, job_name): 

71 try: 

72 operation = self._scheduler.get_job_operation(job_name) 

73 

74 except NotFoundError: 

75 raise InvalidArgumentError(f"Operation name does not exist: [{job_name}]") 

76 

77 metadata = self._scheduler.get_operation_request_metadata(job_name) 

78 return operation, metadata 

79 

80 @DurationMetric(OPERATIONS_LIST_OPERATIONS_TIME_METRIC_NAME, instanced=True) 

81 def list_operations(self, filter_string, page_size, page_token): 

82 if page_size and page_size > self._max_list_operations_page_size: 

83 raise InvalidArgumentError(f"The maximum page size is {self._max_list_operations_page_size}.") 

84 if not page_size: 

85 page_size = self._max_list_operations_page_size 

86 

87 operation_filters = FilterParser.parse_listoperations_filters(filter_string) 

88 if not operation_filters: 

89 operation_filters = DEFAULT_OPERATION_FILTERS 

90 

91 response = operations_pb2.ListOperationsResponse() 

92 

93 results, next_token = self._scheduler.list_operations(operation_filters, page_size, page_token) 

94 response.operations.extend(results) 

95 response.next_page_token = next_token 

96 

97 return response 

98 

99 def delete_operation(self, job_name): 

100 """ DeleteOperation is not supported in BuildGrid. """ 

101 pass 

102 

103 @DurationMetric(OPERATIONS_CANCEL_OPERATION_TIME_METRIC_NAME, instanced=True) 

104 def cancel_operation(self, job_name): 

105 try: 

106 self._scheduler.cancel_job_operation(job_name) 

107 

108 except NotFoundError: 

109 raise InvalidArgumentError(f"Operation name does not exist: [{job_name}]")