TEP-0127: Larger results using sidecar logs - parsing sidecar logs to extract results

internal/sidecarlogresults/sidecarlogresults.go

@@ -17,15 +17,25 @@ limitations under the License.
 package sidecarlogresults
 
 import (
+	"bufio"
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 
+	"github.com/tektoncd/pipeline/pkg/apis/config"
+	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	"golang.org/x/sync/errgroup"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
+// ErrSizeExceeded indicates that the result exceeded its maximum allowed size
+var ErrSizeExceeded = errors.New("results size exceeds configured limit")
+
 // SidecarLogResult holds fields for storing extracted results
 type SidecarLogResult struct {
 	Name  string
@@ -125,3 +135,59 @@ func LookForResults(w io.Writer, runDir string, resultsDir string, resultNames [
 	}
 	return nil
 }
+
+// GetResultsFromSidecarLogs extracts results from the logs of the results sidecar
+func GetResultsFromSidecarLogs(ctx context.Context, clientset kubernetes.Interface, namespace string, name string, container string, podPhase corev1.PodPhase) ([]v1beta1.PipelineResourceResult, error) {
+	sidecarLogResults := []v1beta1.PipelineResourceResult{}
+	if podPhase == corev1.PodPending {
+		return sidecarLogResults, nil
+	}
+	podLogOpts := corev1.PodLogOptions{Container: container}
+	req := clientset.CoreV1().Pods(namespace).GetLogs(name, &podLogOpts)
+	sidecarLogs, err := req.Stream(ctx)
+	if err != nil {
+		return sidecarLogResults, err
+	}
+	defer sidecarLogs.Close()
+	maxResultLimit := config.FromContextOrDefaults(ctx).FeatureFlags.MaxResultSize
+	return extractResultsFromLogs(sidecarLogs, sidecarLogResults, maxResultLimit)
+}
+
+func extractResultsFromLogs(logs io.Reader, sidecarLogResults []v1beta1.PipelineResourceResult, maxResultLimit int) ([]v1beta1.PipelineResourceResult, error) {
+	scanner := bufio.NewScanner(logs)
+	buf := make([]byte, maxResultLimit)
+	scanner.Buffer(buf, maxResultLimit)
+	for scanner.Scan() {
+		result, err := parseResults(scanner.Bytes(), maxResultLimit)
+		if err != nil {
+			return nil, err
+		}
+		sidecarLogResults = append(sidecarLogResults, result)
+	}
+
+	if scanner.Err() != nil {
+		if errors.Is(scanner.Err(), bufio.ErrTooLong) {
+			return sidecarLogResults, ErrSizeExceeded
+		}
+		return nil, scanner.Err()
+	}
+	return sidecarLogResults, nil
+}
+
+func parseResults(resultBytes []byte, maxResultLimit int) (v1beta1.PipelineResourceResult, error) {
+	result := v1beta1.PipelineResourceResult{}
+	if len(resultBytes) > maxResultLimit {
+		return result, ErrSizeExceeded
+	}
+
+	var res SidecarLogResult
+	if err := json.Unmarshal(resultBytes, &res); err != nil {
+		return result, fmt.Errorf("Invalid result %w", err)
+	}
+	result = v1beta1.PipelineResourceResult{
+		Key:        res.Name,
+		Value:      res.Value,
+		ResultType: v1beta1.TaskRunResultType,
+	}
+	return result, nil
+}

internal/sidecarlogresults/sidecarlogresults_test.go

@@ -2,14 +2,22 @@ package sidecarlogresults
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
+	"fmt"
 	"os"
 	"path/filepath"
 	"sort"
+	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	"github.com/tektoncd/pipeline/test/diff"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 )
 
 func TestLookForResults_FanOutAndWait(t *testing.T) {
@@ -121,6 +129,184 @@ func TestLookForResults(t *testing.T) {
 	}
 }
 
+func TestExtractResultsFromLogs(t *testing.T) {
+	inputResults := []SidecarLogResult{
+		{
+			Name:  "result1",
+			Value: "foo",
+		}, {
+			Name:  "result2",
+			Value: "bar",
+		},
+	}
+	podLogs := ""
+	for _, r := range inputResults {
+		res, _ := json.Marshal(&r)
+		podLogs = fmt.Sprintf("%s%s\n", podLogs, string(res))
+	}
+	logs := strings.NewReader(podLogs)
+
+	results, err := extractResultsFromLogs(logs, []v1beta1.PipelineResourceResult{}, 4096)
+	if err != nil {
+		t.Error(err)
+	}
+	want := []v1beta1.PipelineResourceResult{
+		{
+			Key:        "result1",
+			Value:      "foo",
+			ResultType: v1beta1.TaskRunResultType,
+		}, {
+			Key:        "result2",
+			Value:      "bar",
+			ResultType: v1beta1.TaskRunResultType,
+		},
+	}
+	if d := cmp.Diff(want, results); d != "" {
+		t.Fatal(diff.PrintWantGot(d))
+	}
+}
+
+func TestExtractResultsFromLogs_Failure(t *testing.T) {
+	inputResults := []SidecarLogResult{
+		{
+			Name:  "result1",
+			Value: strings.Repeat("v", 4098),
+		},
+	}
+	podLogs := ""
+	for _, r := range inputResults {
+		res, _ := json.Marshal(&r)
+		podLogs = fmt.Sprintf("%s%s\n", podLogs, string(res))
+	}
+	logs := strings.NewReader(podLogs)
+
+	_, err := extractResultsFromLogs(logs, []v1beta1.PipelineResourceResult{}, 4096)
+	if err != ErrSizeExceeded {
+		t.Fatalf("Expected error %v but got %v", ErrSizeExceeded, err)
+	}
+}
+
+func TestParseResults(t *testing.T) {
+	results := []SidecarLogResult{
+		{
+			Name:  "result1",
+			Value: "foo",
+		}, {
+			Name:  "result2",
+			Value: `{"IMAGE_URL":"ar.com", "IMAGE_DIGEST":"sha234"}`,
+		}, {
+			Name:  "result3",
+			Value: `["hello","world"]`,
+		},
+	}
+	podLogs := []string{}
+	for _, r := range results {
+		res, _ := json.Marshal(&r)
+		podLogs = append(podLogs, string(res))
+	}
+	want := []v1beta1.PipelineResourceResult{{
+		Key:        "result1",
+		Value:      "foo",
+		ResultType: v1beta1.TaskRunResultType,
+	}, {
+		Key:        "result2",
+		Value:      `{"IMAGE_URL":"ar.com", "IMAGE_DIGEST":"sha234"}`,
+		ResultType: v1beta1.TaskRunResultType,
+	}, {
+		Key:        "result3",
+		Value:      `["hello","world"]`,
+		ResultType: v1beta1.TaskRunResultType,
+	}}
+	stepResults := []v1beta1.PipelineResourceResult{}
+	for _, plog := range podLogs {
+		res, err := parseResults([]byte(plog), 4096)
+		if err != nil {
+			t.Error(err)
+		}
+		stepResults = append(stepResults, res)
+	}
+	if d := cmp.Diff(want, stepResults); d != "" {
+		t.Fatal(diff.PrintWantGot(d))
+	}
+}
+
+func TestParseResults_Failure(t *testing.T) {
+	result := SidecarLogResult{
+		Name:  "result2",
+		Value: strings.Repeat("k", 4098),
+	}
+	res1, _ := json.Marshal("result1 v1")
+	res2, _ := json.Marshal(&result)
+	podLogs := []string{string(res1), string(res2)}
+	want := []string{
+		"Invalid result json: cannot unmarshal string into Go value of type sidecarlogresults.SidecarLogResult",
+		ErrSizeExceeded.Error(),
+	}
+	got := []string{}
+	for _, plog := range podLogs {
+		_, err := parseResults([]byte(plog), 4096)
+		got = append(got, err.Error())
+	}
+	if d := cmp.Diff(want, got); d != "" {
+		t.Fatal(diff.PrintWantGot(d))
+	}
+}
+
+func TestGetResultsFromSidecarLogs(t *testing.T) {
+	for _, c := range []struct {
+		desc      string
+		podPhase  v1.PodPhase
+		wantError bool
+	}{{
+		desc:      "pod pending to start",
+		podPhase:  corev1.PodPending,
+		wantError: false,
+	}, {
+		desc:      "pod running extract logs",
+		podPhase:  corev1.PodRunning,
+		wantError: true,
+	}} {
+		t.Run(c.desc, func(t *testing.T) {
+			ctx := context.Background()
+			clientset := fakekubeclientset.NewSimpleClientset()
+			pod := &v1.Pod{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Pod",
+					APIVersion: "v1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "pod",
+					Namespace: "foo",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						{
+							Name:  "container",
+							Image: "image",
+						},
+					},
+				},
+				Status: v1.PodStatus{
+					Phase: c.podPhase,
+				},
+			}
+			pod, err := clientset.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+			if err != nil {
+				t.Errorf("Error occurred while creating pod %s: %s", pod.Name, err.Error())
+			}
+
+			// Fake logs are not formatted properly so there will be an error
+			_, err = GetResultsFromSidecarLogs(ctx, clientset, "foo", "pod", "container", pod.Status.Phase)
+			if err != nil && !c.wantError {
+				t.Fatalf("did not expect an error but got: %v", err)
+			}
+			if c.wantError && err == nil {
+				t.Fatal("expected to get an error but did not")
+			}
+		})
+	}
+}
+
 func createResult(t *testing.T, dir string, resultName string, resultValue string) {
 	t.Helper()
 	resultFile := filepath.Join(dir, resultName)

internal/sidecarlogsvalidation/sidecarlogs.go → pkg/apis/pipeline/sidecarlogs.go

@@ -14,10 +14,14 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package sidecarlogsvalidation
+package pipeline
 
 const (
 	// ReservedResultsSidecarName is the name of the results sidecar that outputs the results to stdout
 	// when the results-from feature-flag is set to "sidecar-logs".
 	ReservedResultsSidecarName = "tekton-log-results"
+
+	// ReservedResultsSidecarContainerName is the name of the results sidecar container that is injected
+	// by the reconciler.
+	ReservedResultsSidecarContainerName = "sidecar-tekton-log-results"
 )

pkg/apis/pipeline/v1beta1/taskrun_types.go

@@ -186,6 +186,8 @@ const (
 	TaskRunReasonsResultsVerificationFailed TaskRunReason = "TaskRunResultsVerificationFailed"
 	// AwaitingTaskRunResults is the reason set when waiting upon `TaskRun` results and signatures to verify
 	AwaitingTaskRunResults TaskRunReason = "AwaitingTaskRunResults"
+	// TaskRunReasonResultLargerThanAllowedLimit is the reason set when one of the results exceeds its maximum allowed limit of 1 KB
+	TaskRunReasonResultLargerThanAllowedLimit TaskRunReason = "TaskRunResultLargerThanAllowedLimit"
 )
 
 func (t TaskRunReason) String() string {

pkg/pod/entrypoint.go

@@ -26,6 +26,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/tektoncd/pipeline/pkg/apis/config"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	"gomodules.xyz/jsonpatch/v2"
@@ -252,6 +253,12 @@ func StopSidecars(ctx context.Context, nopImage string, kubeclient kubernetes.In
 	updated := false
 	if newPod.Status.Phase == corev1.PodRunning {
 		for _, s := range newPod.Status.ContainerStatuses {
+			// If the results-from is set to sidecar logs,
+			// a sidecar container with name `sidecar-log-results` is injected by the reconiler.
+			// Do not kill this sidecar. Let it exit gracefully.
+			if config.FromContextOrDefaults(ctx).FeatureFlags.ResultExtractionMethod == config.ResultExtractionMethodSidecarLogs && s.Name == pipeline.ReservedResultsSidecarContainerName {
+				continue
+			}
 			// Stop any running container that isn't a step.
 			// An injected sidecar container might not have the
 			// "sidecar-" prefix, so we can't just look for that