Function streamLogs

cli/cluster/logs.go:74–137

(operatorConfig OperatorConfig, path string, qParams ...map[string]string)

}

func streamLogs(operatorConfig OperatorConfig, path string, qParams ...map[string]string) error {
	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)

	req, err := operatorRequest(operatorConfig, "GET", path, nil, qParams...)
	if err != nil {
		return err
	}

	values := req.URL.Query()
	if operatorConfig.Telemetry {
		values.Set("clientID", operatorConfig.ClientID)
	}

	req.URL.RawQuery = values.Encode()
	wsURL := req.URL.String()
	wsURL = strings.Replace(wsURL, "http", "ws", 1)

	header := http.Header{}
	header.Set("CortexAPIVersion", consts.CortexVersion)
	awsClient, err := aws.New()
	if err != nil {
		return err
	}

	authHeader, err := awsClient.IdentityRequestAsHeader()
	if err != nil {
		return err
	}
	header.Set(consts.AuthHeader, authHeader)

	var dialer = websocket.Dialer{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
	}

	connection, response, err := dialer.Dial(wsURL, header)
	if err != nil && response == nil {
		return ErrorFailedToConnectOperator(err, operatorConfig.EnvName, strings.Replace(operatorConfig.OperatorEndpoint, "http", "ws", 1))
	}
	defer response.Body.Close()

	if err != nil {
		bodyBytes, err := ioutil.ReadAll(response.Body)
		if err != nil || bodyBytes == nil || string(bodyBytes) == "" {
			return ErrorFailedToConnectOperator(err, operatorConfig.EnvName, strings.Replace(operatorConfig.OperatorEndpoint, "http", "ws", 1))
		}
		var output schema.ErrorResponse
		err = json.Unmarshal(bodyBytes, &output)
		if err != nil || output.Message == "" {
			return ErrorOperatorStreamResponseUnknown(string(bodyBytes), response.StatusCode)
		}
		return errors.WithStack(&errors.Error{
			Kind:        output.Kind,
			Message:     output.Message,
			NoTelemetry: true,
		})
	}
	defer connection.Close()