ms/command-executor/src/main/python/utils.py

   1 #
   2 # Copyright (C) 2019 - 2020 Bell Canada.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16 from google.protobuf.timestamp_pb2 import Timestamp
  17
  18 import proto.CommandExecutor_pb2 as CommandExecutor_pb2
  19 import json
  20 import email.parser
  21
  22 CDS_IS_SUCCESSFUL_KEY = "cds_is_successful"
  23 ERR_MSG_KEY = "err_msg"
  24 RESULTS_KEY = "results"
  25 RESULTS_LOG_KEY = "results_log"
  26 REUPLOAD_CBA_KEY = "reupload_cba"
  27 RESPONSE_MAX_SIZE = 4 * 1024 * 1024  # 4Mb
  28
  29 # part of cba_name/version/uuid path
  30 def blueprint_name_version_uuid(request):
  31   return get_blueprint_name(request) + '/' + get_blueprint_version(request) + '/' + get_blueprint_uuid(request)
  32
  33 # return blueprint_name and version part of the path (needed for legacy cmd-exec support
  34 def blueprint_name_version(request):
  35   return get_blueprint_name(request) + '/' + get_blueprint_version(request)
  36
  37 def get_blueprint_name(request):
  38   return request.identifiers.blueprintName
  39
  40 def get_blueprint_version(request):
  41   return request.identifiers.blueprintVersion
  42
  43 def get_blueprint_uuid(request):
  44   return request.identifiers.blueprintUUID
  45
  46 def get_blueprint_timeout(request):
  47   return request.timeOut
  48
  49 def get_blueprint_requestid(request):
  50   return request.requestId
  51
  52 def get_blueprint_subRequestId(request):
  53   return request.subRequestId
  54
  55 # Create a response for grpc. Fills in the timestamp as well as removes cds_is_successful element
  56 def build_grpc_response(request_id, response):
  57   if response[CDS_IS_SUCCESSFUL_KEY]:
  58     status = CommandExecutor_pb2.SUCCESS
  59   else:
  60     status = CommandExecutor_pb2.FAILURE
  61
  62   response.pop(CDS_IS_SUCCESSFUL_KEY)
  63   logs = response.pop(RESULTS_LOG_KEY)
  64
  65   # Payload should only contains response data returned from the executed script and/or the error message
  66   payload = json.dumps(response)
  67
  68   timestamp = Timestamp()
  69   timestamp.GetCurrentTime()
  70
  71   execution_output = CommandExecutor_pb2.ExecutionOutput(requestId=request_id,
  72                                                          response=logs,
  73                                                          status=status,
  74                                                          payload=payload,
  75                                                          timestamp=timestamp)
  76
  77   return truncate_execution_output(execution_output)
  78
  79 def build_grpc_blueprint_upload_response(request_id, subrequest_id, success=True, payload=[]):
  80   timestamp = Timestamp()
  81   timestamp.GetCurrentTime()
  82   return CommandExecutor_pb2.UploadBlueprintOutput(requestId=request_id,
  83     subRequestId=subrequest_id,
  84     status=CommandExecutor_pb2.SUCCESS if success else CommandExecutor_pb2.FAILURE,
  85     timestamp=timestamp,
  86     payload=json.dumps(payload))
  87
  88 # build a ret data structure used to populate the ExecutionOutput
  89 def build_ret_data(cds_is_successful, results_log=[], error=None, reupload_cba = False):
  90   ret_data = {
  91     CDS_IS_SUCCESSFUL_KEY: cds_is_successful,
  92     RESULTS_LOG_KEY: results_log
  93   }
  94   if error:
  95     ret_data[ERR_MSG_KEY] = error
  96   # CBA needs to be reuploaded case:
  97   if reupload_cba:
  98     ret_data[REUPLOAD_CBA_KEY] = True
  99   return ret_data
 100
 101
 102 # Truncate execution logs to make sure gRPC response doesn't exceed the gRPC buffer capacity
 103 def truncate_execution_output(execution_output):
 104   sum_truncated_chars = 0
 105   if execution_output.ByteSize() > RESPONSE_MAX_SIZE:
 106     while execution_output.ByteSize() > RESPONSE_MAX_SIZE:
 107       removed_item = execution_output.response.pop()
 108       sum_truncated_chars += len(removed_item)
 109     execution_output.response.append(
 110         "[...] TRUNCATED CHARS : {}".format(sum_truncated_chars))
 111   return execution_output
 112
 113
 114 # Read temp file 'outputfile' into results_log and split out the returned payload into payload_result
 115 def parse_cmd_exec_output(outputfile, logger, payload_result, results_log,
 116     extra):
 117   payload_section = []
 118   is_payload_section = False
 119   outputfile.seek(0)
 120   while True:
 121     output = outputfile.readline()
 122     if output == '':
 123       break
 124     if output.startswith('BEGIN_EXTRA_PAYLOAD'):
 125       is_payload_section = True
 126       output = outputfile.readline()
 127     if output.startswith('END_EXTRA_PAYLOAD'):
 128       is_payload_section = False
 129       output = ''
 130       payload = '\n'.join(payload_section)
 131       msg = email.parser.Parser().parsestr(payload)
 132       for part in msg.get_payload():
 133         payload_result.update(json.loads(part.get_payload()))
 134     if output and not is_payload_section:
 135       logger.info(output.strip(), extra=extra)
 136       results_log.append(output.strip())
 137     else:
 138       payload_section.append(output.strip())
 139
 140
 141 def getExtraLogData(request=None):
 142   extra = {'request_id': '', 'subrequest_id': '', 'originator_id': ''}
 143   if request is not None:
 144     extra = {
 145       'request_id': request.requestId,
 146       'subrequest_id': request.subRequestId,
 147       'originator_id': request.originatorId
 148     }
 149   return extra