// GetProcessorsLogs retrieves the LSQL processor logs if in kubernetes mode.
GetProcessorsLogs retrieves the LSQL processor logs if in kubernetes mode.
// GetProcessorsLogs retrieves the LSQL processor logs if in kubernetes mode .
func (c *Client) GetProcessorsLogs(clusterName, ns, podName string, follow bool, lines int, handler func(level string, log string) error) error {
	if mode, _ := c.GetExecutionMode(); mode != ExecutionModeKubernetes {
		return fmt.Errorf("unable to retrieve logs, execution mode is not KUBERNETES")
	}
	path := fmt.Sprintf(processorsLogsPathSSE, clusterName, ns, podName)
	if follow {
		if lines <= 0 {
			lines = defaultProcessorsLogsFollowLines
		}
		path += "?follow=true&lines=" + fmt.Sprintf("%d", lines)
	}
	resp, err := c.Do(http.MethodGet, path, contentTypeJSON, nil, func(r *http.Request) error {
		r.Header.Add(acceptHeaderKey, "application/json, text/event-stream")
		return nil
	})
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	reader, err := c.acquireResponseBodyStream(resp)
	if err != nil {
		return err
	}
	streamReader := bufio.NewReader(reader)
	for {
		line, err := streamReader.ReadBytes('\n')
		if err != nil {
			if err == io.EOF {
				return nil 
			}
			return err 
		}
		if len(line) < shiftN+1 { 
			continue
		}
		if !bytes.HasPrefix(line, dataPrefix) {
			continue
		}
		message := line[shiftN:]
		if len(message) < 2 {
			continue
		}
		logEntry := processorLog{}
		if message[0] == '{' {
			if err = json.Unmarshal(message, &logEntry); err == nil {
				t, err := time.Parse(time.RFC3339, logEntry.Timestamp)
				if err == nil {
					logEntry.Timestamp = t.Format("2006-01-02 15:04:05")
				}
				if err = handler(logEntry.Level, fmt.Sprintf("%s %s", logEntry.Timestamp, logEntry.Message)); err != nil {
					return err
				}
			} else {
				handler("info", string(message))
			}
			continue
		}
		handler("", string(message))
	}
}
func ( c * Client ) GetProcessorsLogs ( clusterName , ns , podName string , follow bool , lines int , handler func ( level string , log string ) error ) error { if mode , _ := c . GetExecutionMode ( ) ; mode != ExecutionModeKubernetes { return fmt . Errorf ( " " ) } path := fmt . Sprintf ( processorsLogsPathSSE , clusterName , ns , podName ) if follow { if lines <= 0 { lines = defaultProcessorsLogsFollowLines } path += " " + fmt . Sprintf ( " " , lines ) } resp , err := c . Do ( http . MethodGet , path , contentTypeJSON , nil , func ( r * http . Request ) error { r . Header . Add ( acceptHeaderKey , " " ) return nil } ) if err != nil { return err } defer resp . Body . Close ( ) reader , err := c . acquireResponseBodyStream ( resp ) if err != nil { return err } streamReader := bufio . NewReader ( reader ) for { line , err := streamReader . ReadBytes ( '\n' ) if err != nil { if err == io . EOF { return nil } return err } if len ( line ) < shiftN + 1 { continue } if ! bytes . HasPrefix ( line , dataPrefix ) { continue } message := line [ shiftN : ] if len ( message ) < 2 { continue } logEntry := processorLog { } if message [ 0 ] == '{' { if err = json . Unmarshal ( message , & logEntry ) ; err == nil { t , err := time . Parse ( time . RFC3339 , logEntry . Timestamp ) if err == nil { logEntry . Timestamp = t . Format ( " " ) } if err = handler ( logEntry . Level , fmt . Sprintf ( " " , logEntry . Timestamp , logEntry . Message ) ) ; err != nil { return err } } else { handler ( " " , string ( message ) ) } continue } handler ( " " , string ( message ) ) } }
GetProcessorsLogs retrieves the LSQL processor logs if in kubernetes mode.
GetProcessorsLogs retrieves the LSQL processor logs if in kubernetes mode .
// we read until the the end, exit with no error here. // exit on first failure. // even more +1 for the actual event. // almost empty or totally invalid line, // empty message maybe, // we don't care, we ignore them at any way. // it can be a json object or a pure string log (but always after data:, i.e data:======> Log level set to INFO). // colorized by the caller. // for any case. // it contains the log level itself.