merging again

pkg/kube/pod_logs.go

@@ -33,9 +33,9 @@ type LogOptions struct {
 	// Lines of recent log file to display. Defaults to -1 with no selector, showing all log lines otherwise 10, if a selector is provided.
 	Tail int64
 	// Only return logs after a specific date (RFC3339). Defaults to all logs. Only one of since-time / since may be used.
-	SinceTime time.Time
+	SinceTime *time.Time
 	// Only return logs newer than a relative duration like 5s, 2m, or 3h. Defaults to all logs. Only one of since-time / since may be used.
-	Since time.Duration
+	Since *time.Duration
 	// Print the logs of this container
 	Container string
 	// Selector (label query) to filter on.
@@ -46,7 +46,7 @@ type LogOptions struct {
 	Resource string
 }
 
-func NewOptions() *LogOptions {
+func NewLogOptions() *LogOptions {
 	return &LogOptions{
 		Follow: false,
 		Timestamps: false,
@@ -62,9 +62,15 @@ func NewOptions() *LogOptions {
 	}
 }
 
-func (o *LogOptions) ExecuteLogRequest(out io.Writer) {
+func (o *LogOptions) ExecuteLogRequest(out io.Writer) error {
 	f := cmdutil.NewFactory(nil)
-	Complete(o, f, out)
+	logsOptions, err := Complete(o, f, out)
+	if err != nil {
+		return err
+	}
+	Validate(logsOptions)
+	RunLogs(logsOptions)
+	return nil
 }
 
 func Complete(opts *LogOptions, f cmdutil.Factory, out io.Writer) (*cmd.LogsOptions, error) {
@@ -85,8 +91,8 @@ func Complete(opts *LogOptions, f cmdutil.Factory, out io.Writer) (*cmd.LogsOpti
 		Previous:   opts.Previous,
 		Timestamps: opts.Timestamps,
 	}
-	if opts.SinceTime {
-		t := metav1.NewTime(opts.SinceTime)
+	if opts.SinceTime != nil {
+		t := metav1.NewTime(*opts.SinceTime)
 		logOptions.SinceTime = &t
 	}
 	if opts.LimitBytes != 0 {
@@ -95,7 +101,7 @@ func Complete(opts *LogOptions, f cmdutil.Factory, out io.Writer) (*cmd.LogsOpti
 	if opts.Tail != -1 {
 		logOptions.TailLines = &opts.Tail
 	}
-	if opts.Since {
+	if opts.Since != nil {
 		// round up to the nearest second
 		sec := int64(math.Ceil(opts.Since.Seconds()))
 		logOptions.SinceSeconds = &sec

pkg/tiller/logs/log_streamer.go

@@ -2,6 +2,7 @@ package logs
 
 import (
 	rspb "k8s.io/helm/pkg/proto/hapi/release"
+	"strings"
 )
 
 type Logsub struct {
@@ -11,6 +12,13 @@ type Logsub struct {
 	level   rspb.Log_Level
 }
 
+type LogWriter struct {
+	rls string
+	source rspb.Log_Source
+	level rspb.Log_Level
+	ps *Pubsub
+}
+
 type release struct {
 	name           string
 	sourceMappings map[rspb.Log_Source]map[*Logsub]bool
@@ -85,3 +93,14 @@ func (ps *Pubsub) PubLog(rls string, source rspb.Log_Source, level rspb.Log_Leve
 	}
 }
 
+func (ps *Pubsub) GetWriter(rls string, source rspb.Log_Source, level rspb.Log_Level) *LogWriter {
+	return &LogWriter{rls: rls, source: source, level: level, ps: ps}
+}
+
+func (lw *LogWriter) Write(p []byte) (n int, err error) {
+	logs := strings.Split(string(p), "\n")
+	for _, l := range logs {
+		lw.ps.PubLog(lw.rls, lw.source, lw.level, l)
+	}
+	return len(p), nil
+}

pkg/tiller/release_server.go

@@ -38,7 +38,6 @@ import (
 	"k8s.io/helm/pkg/tiller/environment"
 	"k8s.io/helm/pkg/timeconv"
 	"k8s.io/helm/pkg/version"
-	"k8s.io/helm/pkg/tiller/logs"
 )
 
 // releaseNameMaxLen is the maximum length of a release name.