package pods import ( "context" "io" v1 "k8s.io/api/core/v1" metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" "pandax/apps/devops/services/k8s/logs" ) // maximum number of lines loaded from the apiserver var lineReadLimit int64 = 5000 // maximum number of bytes loaded from the apiserver var byteReadLimit int64 = 500000 // PodContainerList is a list of containers of a pod. type PodContainerList struct { Containers []string `json:"containers"` } // GetPodContainers returns containers that a pod has. func GetPodContainers(client kubernetes.Interface, namespace, podID string) (*PodContainerList, error) { pod, err := client.CoreV1().Pods(namespace).Get(context.TODO(), podID, metaV1.GetOptions{}) if err != nil { return nil, err } containers := &PodContainerList{Containers: make([]string, 0)} for _, container := range pod.Spec.Containers { containers.Containers = append(containers.Containers, container.Name) } return containers, nil } // GetLogDetails returns logs for particular pod and container. When container is null, logs for the first one // are returned. Previous indicates to read archived logs created by log rotation or container crash func GetLogDetails(client kubernetes.Interface, namespace, podID string, container string, logSelector *logs.Selection, usePreviousLogs bool) (*logs.LogDetails, error) { pod, err := client.CoreV1().Pods(namespace).Get(context.TODO(), podID, metaV1.GetOptions{}) if err != nil { return nil, err } if len(container) == 0 { container = pod.Spec.Containers[0].Name } logOptions := mapToLogOptions(container, logSelector, usePreviousLogs) rawLogs, err := readRawLogs(client, namespace, podID, logOptions) if err != nil { return nil, err } details := ConstructLogDetails(podID, rawLogs, container, logSelector) return details, nil } // Maps the log selection to the corresponding api object // Read limits are set to avoid out of memory issues func mapToLogOptions(container string, logSelector *logs.Selection, previous bool) *v1.PodLogOptions { logOptions := &v1.PodLogOptions{ Container: container, Follow: false, Previous: previous, Timestamps: true, } if logSelector.LogFilePosition == logs.Beginning { logOptions.LimitBytes = &byteReadLimit } else { logOptions.TailLines = &lineReadLimit } return logOptions } // Construct a request for getting the logs for a pod and retrieves the logs. func readRawLogs(client kubernetes.Interface, namespace, podID string, logOptions *v1.PodLogOptions) ( string, error) { readCloser, err := openStream(client, namespace, podID, logOptions) if err != nil { return err.Error(), nil } defer readCloser.Close() result, err := io.ReadAll(readCloser) if err != nil { return "", err } return string(result), nil } // GetLogFile returns a stream to the log file which can be piped directly to the response. This avoids out of memory // issues. Previous indicates to read archived logs created by log rotation or container crash func GetLogFile(client kubernetes.Interface, namespace, podID string, container string, opts *v1.PodLogOptions) (io.ReadCloser, error) { logOptions := &v1.PodLogOptions{ Container: container, Follow: false, Previous: opts.Previous, Timestamps: opts.Timestamps, } logStream, err := openStream(client, namespace, podID, logOptions) return logStream, err } func openStream(client kubernetes.Interface, namespace, podID string, logOptions *v1.PodLogOptions) (io.ReadCloser, error) { return client.CoreV1().RESTClient().Get(). Namespace(namespace). Name(podID). Resource("pods"). SubResource("log"). VersionedParams(logOptions, scheme.ParameterCodec).Stream(context.TODO()) } // ConstructLogDetails creates a new log details structure for given parameters. func ConstructLogDetails(podID string, rawLogs string, container string, logSelector *logs.Selection) *logs.LogDetails { parsedLines := logs.ToLogLines(rawLogs) logLines, fromDate, toDate, logSelection, lastPage := parsedLines.SelectLogs(logSelector) readLimitReached := isReadLimitReached(int64(len(rawLogs)), int64(len(parsedLines)), logSelector.LogFilePosition) truncated := readLimitReached && lastPage info := logs.LogInfo{ PodName: podID, ContainerName: container, FromDate: fromDate, ToDate: toDate, Truncated: truncated, } return &logs.LogDetails{ Info: info, Selection: logSelection, LogLines: logLines, } } // Checks if the amount of log file returned from the apiserver is equal to the read limits func isReadLimitReached(bytesLoaded int64, linesLoaded int64, logFilePosition string) bool { return (logFilePosition == logs.Beginning && bytesLoaded >= byteReadLimit) || (logFilePosition == logs.End && linesLoaded >= lineReadLimit) }