-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feature: Logs and Describe overlay (#12)
- Loading branch information
Showing
17 changed files
with
551 additions
and
129 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package k8s | ||
|
|
||
| func SelectorsMatch(first map[string]string, second map[string]string) bool { | ||
| if len(first) != len(second) { | ||
| return false | ||
| } | ||
|
|
||
| for k, v := range first { | ||
| if v2, ok := second[k]; ok { | ||
| if v != v2 { | ||
| return false | ||
| } | ||
| } else { | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| for k2, v2 := range second { | ||
| if v, ok := first[k2]; ok { | ||
| if v2 != v { | ||
| return false | ||
| } | ||
| } else { | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| return true | ||
| } | ||
|
|
||
| func LabelsMatchSelectors(labels map[string]string, selectors map[string]string) bool { | ||
| for k2, v2 := range selectors { | ||
| if v, ok := labels[k2]; ok { | ||
| if v2 != v { | ||
| return false | ||
| } | ||
| } else { | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| return true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| package logs | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "context" | ||
| "encoding/json" | ||
| "strings" | ||
|
|
||
| "github.com/gimlet-io/capacitor/pkg/k8s" | ||
| "github.com/gimlet-io/capacitor/pkg/streaming" | ||
| "github.com/sirupsen/logrus" | ||
| v1 "k8s.io/api/core/v1" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
|
|
||
| "k8s.io/client-go/dynamic" | ||
| "k8s.io/client-go/kubernetes" | ||
| ) | ||
|
|
||
| func Logs( | ||
| client *kubernetes.Clientset, | ||
| dynamicClient *dynamic.DynamicClient, | ||
| namespace string, | ||
| serviceName string, | ||
| clientHub *streaming.ClientHub, | ||
| runningLogStreams *RunningLogStreams, | ||
| ) { | ||
| pods, err := pods(client, namespace, serviceName) | ||
| if err != nil { | ||
| logrus.Warnf("could not get pods to stream logs: %v", err) | ||
| return | ||
| } | ||
|
|
||
| for _, pod := range pods { | ||
| containers := podContainers(pod.Spec) | ||
| for _, container := range containers { | ||
| go streamLogs(client, namespace, pod.Name, container.Name, serviceName, clientHub, runningLogStreams) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func pods(client *kubernetes.Clientset, namespace string, serviceName string) ([]v1.Pod, error) { | ||
| svc, err := client.CoreV1().Services(namespace).Get(context.TODO(), serviceName, metav1.GetOptions{}) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| podsInNamespace, err := client.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{}) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| pods := []v1.Pod{} | ||
| for _, pod := range podsInNamespace.Items { | ||
| if k8s.LabelsMatchSelectors(pod.ObjectMeta.Labels, svc.Spec.Selector) { | ||
| pods = append(pods, pod) | ||
| } | ||
| } | ||
|
|
||
| return pods, nil | ||
| } | ||
|
|
||
| func podContainers(podSpec v1.PodSpec) (containers []v1.Container) { | ||
| containers = append(containers, podSpec.InitContainers...) | ||
| containers = append(containers, podSpec.Containers...) | ||
|
|
||
| return containers | ||
| } | ||
|
|
||
| func streamLogs( | ||
| client *kubernetes.Clientset, | ||
| namespace string, | ||
| pod string, | ||
| containerName string, | ||
| serviceName string, | ||
| clientHub *streaming.ClientHub, | ||
| runningLogStreams *RunningLogStreams, | ||
| ) { | ||
| count := int64(100) | ||
| podLogOpts := v1.PodLogOptions{ | ||
| Container: containerName, | ||
| TailLines: &count, | ||
| Follow: true, | ||
| Timestamps: true, | ||
| } | ||
| logsReq := client.CoreV1().Pods(namespace).GetLogs(pod, &podLogOpts) | ||
|
|
||
| podLogs, err := logsReq.Stream(context.Background()) | ||
| if err != nil { | ||
| logrus.Errorf("could not stream pod logs: %v", err) | ||
| return | ||
| } | ||
| defer podLogs.Close() | ||
|
|
||
| stopCh := runningLogStreams.register(namespace, serviceName) | ||
|
|
||
| go func() { | ||
| <-stopCh | ||
| podLogs.Close() | ||
| }() | ||
|
|
||
| sc := bufio.NewScanner(podLogs) | ||
| for sc.Scan() { | ||
| text := sc.Text() | ||
| chunks := chunks(text, 1000) | ||
| for _, chunk := range chunks { | ||
| timestamp, message := parseMessage(chunk) | ||
| payload := streaming.PodLogMessage{ | ||
| Timestamp: timestamp, | ||
| Container: containerName, | ||
| Pod: pod, | ||
| Svc: namespace + "/" + serviceName, | ||
| Message: message, | ||
| } | ||
|
|
||
| msgBytes, err := json.Marshal(streaming.Envelope{ | ||
| Type: streaming.POD_LOGS_RECEIVED, | ||
| Payload: payload, | ||
| }) | ||
|
|
||
| if err != nil { | ||
| logrus.Error("cannot serialize message", err) | ||
| } | ||
|
|
||
| clientHub.Broadcast <- msgBytes | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func chunks(str string, size int) []string { | ||
| if len(str) <= size { | ||
| return []string{str} | ||
| } | ||
| return append([]string{string(str[0:size])}, chunks(str[size:], size)...) | ||
| } | ||
|
|
||
| func parseMessage(chunk string) (string, string) { | ||
| parts := strings.SplitN(chunk, " ", 2) | ||
|
|
||
| return parts[0], parts[1] | ||
| } |
Oops, something went wrong.