Update k8s log options

debug/log/kubernetes/kubernetes.go

@@ -24,7 +24,12 @@ func (k *klog) podLogStream(podName string, stream *kubeStream) {
 	p := make(map[string]string)
 	p["follow"] = "true"
 
-	body, err := k.client.Logs(podName, client.AdditionalParams(p))
+	// get the logs for the pod
+	body, err := k.client.Log(&client.Resource{
+		Name: podName,
+		Kind: "pod",
+	}, client.LogParams(p))
+
 	if err != nil {
 		fmt.Fprintf(os.Stderr, err.Error())
 		return
@@ -98,14 +103,14 @@ func (k *klog) Read(options ...log.ReadOption) ([]log.Record, error) {
 		o(opts)
 	}
 
-	logsToGet, err := k.getMatchingPods()
+	pods, err := k.getMatchingPods()
 	if err != nil {
 		return nil, err
 	}
 
 	var records []log.Record
 
-	for _, l := range logsToGet {
+	for _, pod := range pods {
 		logParams := make(map[string]string)
 
 		if !opts.Since.Equal(time.Time{}) {
@@ -120,7 +125,11 @@ func (k *klog) Read(options ...log.ReadOption) ([]log.Record, error) {
 			logParams["follow"] = "true"
 		}
 
-		logs, err := k.client.Logs(l, client.AdditionalParams(logParams))
+		logs, err := k.client.Log(&client.Resource{
+			Name: pod,
+			Kind: "pod",
+		}, client.LogParams(logParams))
+
 		if err != nil {
 			return nil, err
 		}
@@ -130,7 +139,7 @@ func (k *klog) Read(options ...log.ReadOption) ([]log.Record, error) {
 
 		for s.Scan() {
 			record := k.parse(s.Text())
-			record.Metadata["pod"] = l
+			record.Metadata["pod"] = pod
 			records = append(records, record)
 		}
 	}

util/kubernetes/client/client.go

@@ -138,20 +138,21 @@ func (c *client) Get(r *Resource, labels map[string]string) error {
 		Into(r.Value)
 }
 
-// Logs returns logs for a pod
+// Log returns logs for a pod
-func (c *client) Logs(podName string, options ...LogOption) (io.ReadCloser, error) {
+func (c *client) Log(r *Resource, opts ...LogOption) (io.ReadCloser, error) {
-	opts := &LogOptions{}
+	var options LogOptions
-	for _, o := range options {
+	for _, o := range opts {
-		o(opts)
+		o(&options)
 	}
+
 	req := api.NewRequest(c.opts).
 		Get().
-		Resource("pod").
+		Resource(r.Kind).
 		SubResource("log").
-		Name(podName)
+		Name(r.Name)
 
-	if opts.AdditionalParams != nil {
+	if options.Params != nil {
-		req.Params(&api.Params{Additional: opts.AdditionalParams})
+		req.Params(&api.Params{Additional: options.Params})
 	}
 
 	resp, err := req.Raw()

util/kubernetes/client/kubernetes.go

@@ -25,8 +25,8 @@ type Kubernetes interface {
 	Delete(*Resource) error
 	// List lists API resources
 	List(*Resource) error
-	// Logs gets logs from a pod
+	// Log gets log for a pod
-	Logs(string, ...LogOption) (io.ReadCloser, error)
+	Log(*Resource, ...LogOption) (io.ReadCloser, error)
 }
 
 // NewService returns default micro kubernetes service definition

util/kubernetes/client/logoptions.go

@@ -1,13 +0,0 @@
-package client
-
-type LogOptions struct {
-	AdditionalParams map[string]string
-}
-
-type LogOption func(*LogOptions)
-
-func AdditionalParams(p map[string]string) LogOption {
-	return func(l *LogOptions) {
-		l.AdditionalParams = p
-	}
-}

util/kubernetes/client/options.go

@@ -0,0 +1,14 @@
+package client
+
+type LogOptions struct {
+	Params map[string]string
+}
+
+type LogOption func(*LogOptions)
+
+// LogParams provides additional params for logs
+func LogParams(p map[string]string) LogOption {
+	return func(l *LogOptions) {
+		l.Params = p
+	}
+}