add log client (#947) (#950)

* add log client

Author: zhongzichao

client/paddleflow/cli/log.py

@@ -64,6 +64,43 @@ def show(ctx, runid, jobid=None, pagesize=None, pageno=None, logfileposition=Non
         sys.exit(1)
 
 
+@log.command(context_settings=dict(max_content_width=2000), cls=command_required_option_from_option())
+@click.option('-j', '--jobid', help="jobid and name is one of required field")
+@click.option('-n', '--name', help="name")
+@click.option('-ns', '--namespace', help="namespace")
+@click.option('-c', '--clustername', help="clustername")
+@click.option('-r', '--readfromtail', help="read logs from tail, set value means yes")
+@click.option('-l', '--line_limit', help="line_limit, default is 1000")
+@click.option('-s', '--size_limit', help="size_limit, default is 100MB")
+@click.option('-t', '--type', help="type in deploy or pod")
+@click.option('-f', '--framework', help="job framework")
+@click.pass_context
+def showlimit(ctx, jobid=None, name=None, namespace=None, clustername=None, readfromtail=None, line_limit=None, size_limit=None, type=None, framework=None):
+    """
+    show job,deployment,pod log\n
+    if jobid is null, it would return deployment or pod logs by <namespace, name, type>
+    else it would return PF job logs by jobid
+
+    When both A and B have values, the minimum value is selected to return to the log
+    """
+    client = ctx.obj['client']
+    output_format = 'text'
+    if jobid:
+        click.echo("query job by [jobid]".format(jobid))
+    elif name:
+        click.echo("query log by [namespace/name]".format(namespace, name))
+
+    valid, response = client.show_log_by_limit(job_id=jobid, name=name, namespace=namespace, cluster_name=clustername, read_from_tail=readfromtail,
+                                               line_limit=line_limit, size_limit=size_limit, type=type, framework=framework)
+    if valid:
+        # if jobid is None and len(response['runLog']) > 0:
+        #     response['runLog'] = [response['runLog'][0]]
+        _print_log_by_limit(response, output_format)
+    else:
+        click.echo("show logs failed with message[%s]" % response)
+        sys.exit(1)
+
+
 def _print_run_log(loginfo, out_format):
     """print run log """
     submit_loginfo = loginfo['submitLog']
@@ -84,3 +121,35 @@ def _print_run_log(loginfo, out_format):
         if index != len(run_loginfo) - 1:
             data.append(['\n'])
     print_output(data, [], out_format)
+
+def _print_log_by_limit(loginfo, out_format):
+    """print run log by limit """
+    eventList = loginfo['eventList']
+    logs = loginfo['logs']
+
+    if eventList is None or len(eventList) == 0:
+        click.echo(click.style('eventList is None', fg='green'))
+    else:
+        click.echo(click.style('eventList:', fg='green'))
+        for index, item in enumerate(eventList):
+            click.echo(item)
+
+    click.echo(click.style('logs:', fg='green'))
+
+    data = []
+    if len(logs) == 0:
+        data.append(["None"])
+    for index, item in enumerate(logs):
+        if item.job_id:
+            line = ["jobid:" + item.job_id, "taskid:" + item.task_id, "has_next_page:" + str(item.has_next_page),
+                    "truncated:" + str(item.truncated)]
+        else:
+            line = ["name:" + item.name, "taskid:" + item.task_id, "has_next_page:" + str(item.has_next_page),
+                    "line_limit:" + str(item.line_limit), "size_limit:" + str(item.size_limit), "truncated:" + str(item.truncated)]
+        line_log = []
+        line_log.append(item.log_content)
+        data.append(line)
+        data.append(line_log)
+        if index != len(logs) - 1:
+            data.append(['\n'])
+    print_output(data, [], out_format)

client/paddleflow/client.py

@@ -714,6 +714,19 @@ def show_log(self, run_id, job_id=None, page_size=None, page_no=None, log_file_p
         return LogServiceApi.get_log_info(self.paddleflow_server, run_id, job_id, page_size, page_no, log_file_position,
                                           self.header)
 
+    def show_log_by_limit(self, job_id=None, name=None, namespace=None, cluster_name=None, read_from_tail=None,
+                          line_limit=None, size_limit=None,
+                          type=None, framework=None):
+        """
+        show job logs or kubernetes logs or others
+        """
+        self.pre_check()
+        return LogServiceApi.get_log_info_by_limit(self.paddleflow_server, job_id=job_id, name=name, namespace=namespace,
+                               cluster_name=cluster_name, read_from_tail=read_from_tail, line_limit=line_limit, size_limit=size_limit,
+                                                   type=type, framework=framework, header=self.header
+                                                   # job_id=job_id,
+                                                   )
+
     def create_job(self, job_type, job_request):
         """
         create_job

client/paddleflow/common/api.py

@@ -33,7 +33,7 @@
 PADDLE_FLOW_ARTIFACT = '/api/paddleflow/v%d/artifact' % PADDLE_FLOW_VERSION
 PADDLE_FLOW_SCHEDULE = '/api/paddleflow/v%d/schedule' % PADDLE_FLOW_VERSION
 PADDLE_FLOW_FLAVOUR = '/api/paddleflow/v%d/flavour' % PADDLE_FLOW_VERSION
-PADDLE_FLOW_LOG = '/api/paddleflow/v%d/log/run' % PADDLE_FLOW_VERSION
+PADDLE_FLOW_LOG = '/api/paddleflow/v%d/log' % PADDLE_FLOW_VERSION
 PADDLE_FLOW_JOB = '/api/paddleflow/v%d/job' % PADDLE_FLOW_VERSION
 PADDLE_FLOW_STATISTIC = '/api/paddleflow/v%d/statistics' % PADDLE_FLOW_VERSION
 PADDLE_FLOW_SERVER_VERSION = '/api/paddleflow/v%d/version' % PADDLE_FLOW_VERSION

client/paddleflow/log/log_api.py

@@ -21,11 +21,13 @@
 
 from paddleflow.common import api
 from paddleflow.common.exception.paddleflow_sdk_exception import PaddleFlowSDKException
-from paddleflow.log.log_info import LogInfo
+from paddleflow.log.log_info import LogInfo, LogInfoByLimit
 from paddleflow.utils import api_client
 
 DEFAULT_PAGESIZE = 100
 DEFAULT_PAGENO = 1
+DEFAULT_LINE_LIMIT = 1000
+DEFAULT_SIZE_LIMIT = "100KB"
 
 class LogServiceApi(object):
     """
@@ -51,7 +53,7 @@ def get_log_info(self, host, runid, jobid, pagesize, pageno, fileposition, heade
             params['pageNo'] = pageno
         if fileposition:
             params['logFilePosition'] = fileposition
-        response = api_client.call_api(method="GET", url=parse.urljoin(host, api.PADDLE_FLOW_LOG + "/%s" % runid),
+        response = api_client.call_api(method="GET", url=parse.urljoin(host, api.PADDLE_FLOW_LOG + "/run/%s" % runid),
                                        headers=header, params=params)
         if not response:
             raise PaddleFlowSDKException("Connection Error", "get log failed due to HTTPError")
@@ -77,3 +79,58 @@ def get_log_info(self, host, runid, jobid, pagesize, pageno, fileposition, heade
         return True, log_info_dict
 
 
+    @classmethod
+    def get_log_info_by_limit(self, host, job_id=None, name=None, namespace=None,
+                              cluster_name=None, read_from_tail=None, line_limit=None, size_limit=None,
+                              type=None, framework=None, header=None):
+        """ get logs by line_limit or size_limit
+        """
+        if not header:
+            raise PaddleFlowSDKException("InvalidRequest", "paddleflow should login first")
+        params = {}
+        if job_id:
+            params['jobID'] = job_id
+        if name:
+            params['name'] = name
+        if namespace:
+            params['namespace'] = namespace
+        if cluster_name:
+            params['clusterName'] = cluster_name
+        if read_from_tail:
+            params['readFromTail'] = read_from_tail
+        if line_limit:
+            params['lineLimit'] = str(line_limit)
+        if size_limit:
+            params['sizeLimit'] = str(size_limit)
+        if type:
+            params['type'] = type
+        if framework:
+            params['framework'] = framework
+
+        response = api_client.call_api(method="GET", url=parse.urljoin(host, api.PADDLE_FLOW_LOG + "/job"),
+                                       headers=header, params=params)
+        if not response:
+            raise PaddleFlowSDKException("Connection Error", "get log failed due to HTTPError")
+        data = json.loads(response.text)
+        if 'message' in data:
+            return False, data['message']
+        if line_limit is None:
+            line_limit = DEFAULT_LINE_LIMIT
+        else:
+            line_limit = int(line_limit)
+        if size_limit is None:
+            size_limit = DEFAULT_SIZE_LIMIT
+
+
+        loginfo_list = []
+        for task in data['taskList']:
+            loginfo = LogInfoByLimit(job_id=data['jobID'], task_id=task['taskID'], name=data['name'],
+                                     has_next_page=task['logInfo']['hasNextPage'],
+                                     truncated=task['logInfo']['truncated'],
+                                     line_limit=line_limit, size_limit=size_limit, log_content=task['logInfo']['logContent'])
+            loginfo_list.append(loginfo)
+
+        log_info_dict = {'eventList': data['eventList'], 'logs': loginfo_list}
+        return True, log_info_dict
+
+

client/paddleflow/log/log_info.py

@@ -39,3 +39,26 @@ def __init__(self, runid, jobid, taskid, has_next_page, truncated, pagesize, pag
         self.pageno = pageno
         # 具体的日志内容
         self.log_content = log_content
+
+
+class LogInfoByLimit(object):
+
+    """the class of log info"""
+    def __init__(self, job_id="", task_id="", name="", has_next_page=False, truncated=False, line_limit=0, size_limit="", log_content=""):
+        """init """
+        # job id
+        self.job_id = job_id
+        # job下子task的id
+        self.task_id = task_id
+        # name
+        self.name = name
+        # 日志内容是否还有下一页，为true时则有下一页，否则为最后一页
+        self.has_next_page = has_next_page
+        # 日志内容是否被截断，为true时则被截断，否则未截断
+        self.truncated = truncated
+        # 行数限制
+        self.line_limit = line_limit
+        # 字节数限制
+        self.size_limit = size_limit
+        # 具体的日志内容
+        self.log_content = log_content