Merge pull request #108 from xuzhu-591/remove-s3

Remove s3

.github/workflows/image.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       matrix:
         components: [core, swagger]
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v3
         with:

.github/workflows/ut.yml

@@ -18,7 +18,7 @@ permissions:
 jobs:
   lint:
     name: golangci-lint
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/setup-go@v3
         with:
@@ -35,7 +35,7 @@ jobs:
         go: [1.15]
     name: unit-test
     needs: [lint]
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/setup-go@v3
         with:

config.yaml

@@ -50,7 +50,11 @@ tektonMapper:
     # if you run horizon on local machine, you need to set this to the absolute path of your kubeconfig
     # if you are running it on mac, the path may be looked like '/Users/xxx/.kube/config'
     kubeconfig: "/Users/xxx/.kube/config"
-    s3:
+    logStorage:
+      # the following types of log storage are supported:
+      #   s3: Minio is used by default, and you can also specify your own s3 storage.
+      #   dummy: A dummy log storage. Building logs are not stored and Horizon gets them directly from k8s.
+      type: dummy
       accessKey: ""
       secretKey: ""
       region: ""

core/controller/cloudevent/controller.go

@@ -92,20 +92,6 @@ func (c *controller) CloudEvent(ctx context.Context, wpr *WrappedPipelineRun) (e
 		return err
 	}
 
-	tekton, err := c.tektonFty.GetTekton(environment)
-	if err != nil {
-		return err
-	}
-
-	// 3. delete pipelinerun in k8s
-	if err := tekton.DeletePipelineRun(ctx, wpr.PipelineRun); err != nil {
-		if _, ok := perror.Cause(err).(*herrors.HorizonErrNotFound); ok {
-			log.Warningf(ctx, "received pipelineRun: %v is not found when delete", wpr.PipelineRun.Name)
-			return nil
-		}
-		return err
-	}
-
 	// format Pipeline results
 	pipelineResult := metrics.FormatPipelineResults(wpr.PipelineRun)
 

core/controller/cloudevent/controller_test.go

@@ -233,8 +233,6 @@ func Test(t *testing.T) {
 		}(),
 	}, nil)
 
-	tekton.EXPECT().DeletePipelineRun(ctx, gomock.Any()).Return(nil)
-
 	templateReleaseMgr := trmock.NewMockManager(mockCtl)
 	templateReleaseMgr.EXPECT().GetByTemplateNameAndRelease(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(&trmodels.TemplateRelease{}, nil)

core/controller/cluster/controller_status.go

@@ -253,41 +253,11 @@ func (c *controller) getLatestPipelinerunByClusterID(ctx context.Context,
 
 func (c *controller) getLatestPipelineRunObject(ctx context.Context, cluster *clustermodels.Cluster,
 	pipelinerun *prmodels.Pipelinerun) (*v1beta1.PipelineRun, error) {
-	var latestPr *v1beta1.PipelineRun
-	getPipelineRunFromCollector := func() (*v1beta1.PipelineRun, error) {
-		tektonCollector, err := c.tektonFty.GetTektonCollector(cluster.EnvironmentName)
-		if err != nil {
-			return nil, err
-		}
-		obj, err := tektonCollector.GetPipelineRunObject(ctx, pipelinerun.PrObject)
-		if err != nil {
-			return nil, err
-		}
-		return obj.PipelineRun, nil
-	}
-	var (
-		err          error
-		tektonClient tekton.Interface
-	)
-	if pipelinerun.PrObject == "" {
-		tektonClient, err = c.tektonFty.GetTekton(cluster.EnvironmentName)
-		if err != nil {
-			return nil, err
-		}
-		latestPr, err = tektonClient.GetPipelineRunByID(ctx, pipelinerun.CIEventID)
-		if err != nil {
-			if _, ok := perror.Cause(err).(*herrors.HorizonErrNotFound); ok {
-				return nil, nil
-			}
-			return nil, err
-		}
-	} else {
-		latestPr, err = getPipelineRunFromCollector()
-		if err != nil {
-			return nil, err
-		}
+	tektonCollector, err := c.tektonFty.GetTektonCollector(cluster.EnvironmentName)
+	if err != nil {
+		return nil, err
 	}
-	return latestPr, nil
+	return tektonCollector.GetPipelineRun(ctx, pipelinerun)
 }
 
 // getRunningTask Get the latest currently executing Task of the pipeline Run.

core/controller/cluster/controller_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	tektoncollectormock "github.com/horizoncd/horizon/mock/pkg/cluster/tekton/collector"
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/horizoncd/horizon/core/common"
@@ -802,6 +803,10 @@ func test(t *testing.T) {
 	tektonFty.EXPECT().GetTekton(gomock.Any()).Return(tekton, nil).AnyTimes()
 	tekton.EXPECT().CreatePipelineRun(ctx, gomock.Any()).Return("abc", nil)
 	tekton.EXPECT().GetPipelineRunByID(ctx, gomock.Any()).Return(pr, nil).AnyTimes()
+	tektonCollector := tektoncollectormock.NewMockInterface(mockCtl)
+
+	tektonFty.EXPECT().GetTektonCollector(gomock.Any()).Return(tektonCollector, nil).AnyTimes()
+	tektonCollector.EXPECT().GetPipelineRun(ctx, gomock.Any()).Return(pr, nil).AnyTimes()
 
 	commitGetter.EXPECT().GetCommit(ctx, gomock.Any(), gomock.Any(), gomock.Any()).Return(&git.Commit{
 		ID:      "code-commit-id",

core/controller/pipelinerun/controller.go

@@ -11,8 +11,8 @@ import (
 	codemodels "github.com/horizoncd/horizon/pkg/cluster/code"
 	"github.com/horizoncd/horizon/pkg/cluster/gitrepo"
 	clustermanager "github.com/horizoncd/horizon/pkg/cluster/manager"
+	"github.com/horizoncd/horizon/pkg/cluster/tekton/collector"
 	"github.com/horizoncd/horizon/pkg/cluster/tekton/factory"
-	"github.com/horizoncd/horizon/pkg/cluster/tekton/log"
 	envmanager "github.com/horizoncd/horizon/pkg/environment/manager"
 	perror "github.com/horizoncd/horizon/pkg/errors"
 	"github.com/horizoncd/horizon/pkg/param"
@@ -25,8 +25,8 @@ import (
 )
 
 type Controller interface {
-	GetPipelinerunLog(ctx context.Context, pipelinerunID uint) (*Log, error)
-	GetClusterLatestLog(ctx context.Context, clusterID uint) (*Log, error)
+	GetPipelinerunLog(ctx context.Context, pipelinerunID uint) (*collector.Log, error)
+	GetClusterLatestLog(ctx context.Context, clusterID uint) (*collector.Log, error)
 	GetDiff(ctx context.Context, pipelinerunID uint) (*GetDiffResponse, error)
 	Get(ctx context.Context, pipelinerunID uint) (*PipelineBasic, error)
 	List(ctx context.Context, clusterID uint, canRollback bool, query q.Query) (int, []*PipelineBasic, error)
@@ -60,14 +60,7 @@ func NewController(param *param.Param) Controller {
 	}
 }
 
-type Log struct {
-	LogChannel <-chan log.Log
-	ErrChannel <-chan error
-
-	LogBytes []byte
-}
-
-func (c *controller) GetPipelinerunLog(ctx context.Context, pipelinerunID uint) (_ *Log, err error) {
+func (c *controller) GetPipelinerunLog(ctx context.Context, pipelinerunID uint) (_ *collector.Log, err error) {
 	const op = "pipelinerun controller: get pipelinerun log"
 	defer wlog.Start(ctx, op).StopPrint()
 
@@ -89,7 +82,7 @@ func (c *controller) GetPipelinerunLog(ctx context.Context, pipelinerunID uint)
 	return c.getPipelinerunLog(ctx, pr, cluster.EnvironmentName)
 }
 
-func (c *controller) GetClusterLatestLog(ctx context.Context, clusterID uint) (_ *Log, err error) {
+func (c *controller) GetClusterLatestLog(ctx context.Context, clusterID uint) (_ *collector.Log, err error) {
 	const op = "pipelinerun controller: get cluster latest log"
 	defer wlog.Start(ctx, op).StopPrint()
 
@@ -109,39 +102,16 @@ func (c *controller) GetClusterLatestLog(ctx context.Context, clusterID uint) (_
 }
 
 func (c *controller) getPipelinerunLog(ctx context.Context, pr *prmodels.Pipelinerun,
-	environment string) (_ *Log, err error) {
+	environment string) (_ *collector.Log, err error) {
 	const op = "pipeline controller: get pipelinerun log"
 	defer wlog.Start(ctx, op).StopPrint()
 
-	// if pr.PrObject is empty, get pipelinerun log in k8s
-	if pr.PrObject == "" {
-		tektonClient, err := c.tektonFty.GetTekton(environment)
-		if err != nil {
-			return nil, perror.WithMessagef(err, "faild to get tekton for %s", environment)
-		}
-
-		logCh, errCh, err := tektonClient.GetPipelineRunLogByID(ctx, pr.CIEventID)
-		if err != nil {
-			return nil, err
-		}
-		return &Log{
-			LogChannel: logCh,
-			ErrChannel: errCh,
-		}, nil
-	}
-
-	// else, get log from s3
 	tektonCollector, err := c.tektonFty.GetTektonCollector(environment)
 	if err != nil {
 		return nil, perror.WithMessagef(err, "faild to get tekton collector for %s", environment)
 	}
-	logBytes, err := tektonCollector.GetPipelineRunLog(ctx, pr.LogObject)
-	if err != nil {
-		return nil, perror.WithMessagef(err, "faild to get tekton collector for %s", environment)
-	}
-	return &Log{
-		LogBytes: logBytes,
-	}, nil
+
+	return tektonCollector.GetPipelineRunLog(ctx, pr)
 }
 
 func (c *controller) GetDiff(ctx context.Context, pipelinerunID uint) (_ *GetDiffResponse, err error) {

core/controller/pipelinerun/controller_test.go

@@ -24,6 +24,7 @@ import (
 	userauth "github.com/horizoncd/horizon/pkg/authentication/user"
 	codemodels "github.com/horizoncd/horizon/pkg/cluster/code"
 	clustermodel "github.com/horizoncd/horizon/pkg/cluster/models"
+	"github.com/horizoncd/horizon/pkg/cluster/tekton/collector"
 	"github.com/horizoncd/horizon/pkg/cluster/tekton/log"
 	envmodels "github.com/horizoncd/horizon/pkg/environmentregion/models"
 	"github.com/horizoncd/horizon/pkg/git"
@@ -270,11 +271,8 @@ func Test(t *testing.T) {
 	}
 
 	logBytes := []byte("this is a log")
-	tektonCollector.EXPECT().GetPipelineRunLog(ctx, gomock.Any()).Return(logBytes, nil).AnyTimes()
-
-	logCh := make(chan log.Log)
-	errCh := make(chan error)
-	tekton.EXPECT().GetPipelineRunLogByID(ctx, gomock.Any()).Return(logCh, errCh, nil)
+	tektonCollector.EXPECT().GetPipelineRunLog(ctx, gomock.Any()).
+		Return(&collector.Log{LogBytes: logBytes}, nil).Times(1)
 
 	l, err := c.GetPipelinerunLog(ctx, pipelinerun.ID)
 	assert.Nil(t, err)
@@ -295,6 +293,14 @@ func Test(t *testing.T) {
 	assert.Nil(t, err)
 	assert.NotNil(t, pipelinerun)
 
+	logCh := make(chan log.Log)
+	errCh := make(chan error)
+	tektonCollector.EXPECT().GetPipelineRunLog(ctx, gomock.Any()).
+		Return(&collector.Log{
+			LogChannel: logCh,
+			ErrChannel: errCh,
+		}, nil).Times(1)
+
 	go func() {
 		defer close(logCh)
 		defer close(errCh)

core/http/api/v1/pipelinerun/apis.go

@@ -7,6 +7,7 @@ import (
 	"github.com/horizoncd/horizon/core/common"
 	prctl "github.com/horizoncd/horizon/core/controller/pipelinerun"
 	"github.com/horizoncd/horizon/lib/q"
+	"github.com/horizoncd/horizon/pkg/cluster/tekton/collector"
 	"github.com/horizoncd/horizon/pkg/server/request"
 	"github.com/horizoncd/horizon/pkg/server/response"
 	"github.com/horizoncd/horizon/pkg/util/errors"
@@ -40,7 +41,7 @@ func (a *API) Log(c *gin.Context) {
 	}
 	l, err := a.prCtl.GetPipelinerunLog(c, uint(prID))
 	if err != nil {
-		l := &prctl.Log{
+		l := &collector.Log{
 			LogBytes: []byte(errors.Message(err)),
 		}
 		a.writeLog(c, l)
@@ -64,7 +65,7 @@ func (a *API) LatestLogForCluster(c *gin.Context) {
 	a.writeLog(c, l)
 }
 
-func (a *API) writeLog(c *gin.Context, l *prctl.Log) {
+func (a *API) writeLog(c *gin.Context, l *collector.Log) {
 	c.Header("Content-Type", "text/plain")
 	if l.LogBytes != nil {
 		_, _ = c.Writer.Write(l.LogBytes)

core/http/api/v1/pipelinerun/routers.go

@@ -10,30 +10,30 @@ import (
 )
 
 // RegisterRoutes register routes
-func (api *API) RegisterRoute(engine *gin.Engine) {
+func (a *API) RegisterRoute(engine *gin.Engine) {
 	apiGroup := engine.Group("/apis/core/v1")
 	var routes = route.Routes{
 		{
 			Method:      http.MethodGet,
 			Pattern:     fmt.Sprintf("/pipelineruns/:%v/log", _pipelinerunIDParam),
-			HandlerFunc: api.Log,
+			HandlerFunc: a.Log,
 		}, {
 			Method:      http.MethodPost,
 			Pattern:     fmt.Sprintf("/pipelineruns/:%v/stop", _pipelinerunIDParam),
-			HandlerFunc: api.Stop,
+			HandlerFunc: a.Stop,
 		}, {
 			Method:      http.MethodGet,
 			Pattern:     fmt.Sprintf("/pipelineruns/:%v/diffs", _pipelinerunIDParam),
-			HandlerFunc: api.GetDiff,
+			HandlerFunc: a.GetDiff,
 		}, {
 			Method:      http.MethodGet,
 			Pattern:     fmt.Sprintf("/pipelineruns/:%v", _pipelinerunIDParam),
-			HandlerFunc: api.Get,
+			HandlerFunc: a.Get,
 		},
 		{
 			Method:      http.MethodGet,
 			Pattern:     fmt.Sprintf("/clusters/:%v/pipelineruns", _clusterIDParam),
-			HandlerFunc: api.List,
+			HandlerFunc: a.List,
 		},
 	}
 
@@ -45,11 +45,11 @@ func (api *API) RegisterRoute(engine *gin.Engine) {
 		{
 			Method:      http.MethodGet,
 			Pattern:     fmt.Sprintf("/clusters/:%v/log", _clusterParam),
-			HandlerFunc: api.LatestLogForCluster,
+			HandlerFunc: a.LatestLogForCluster,
 		}, {
 			Method:      http.MethodPost,
 			Pattern:     fmt.Sprintf("/clusters/:%v/stop", _clusterParam),
-			HandlerFunc: api.StopPipelinerunForCluster,
+			HandlerFunc: a.StopPipelinerunForCluster,
 		},
 	}
 

core/http/api/v2/pipelinerun/apis.go

@@ -7,6 +7,7 @@ import (
 	"github.com/horizoncd/horizon/core/common"
 	prctl "github.com/horizoncd/horizon/core/controller/pipelinerun"
 	"github.com/horizoncd/horizon/lib/q"
+	"github.com/horizoncd/horizon/pkg/cluster/tekton/collector"
 	"github.com/horizoncd/horizon/pkg/server/request"
 	"github.com/horizoncd/horizon/pkg/server/response"
 	"github.com/horizoncd/horizon/pkg/util/errors"
@@ -39,7 +40,7 @@ func (a *API) Log(c *gin.Context) {
 	}
 	l, err := a.prCtl.GetPipelinerunLog(c, uint(prID))
 	if err != nil {
-		l := &prctl.Log{
+		l := &collector.Log{
 			LogBytes: []byte(errors.Message(err)),
 		}
 		a.writeLog(c, l)
@@ -48,7 +49,7 @@ func (a *API) Log(c *gin.Context) {
 	a.writeLog(c, l)
 }
 
-func (a *API) writeLog(c *gin.Context, l *prctl.Log) {
+func (a *API) writeLog(c *gin.Context, l *collector.Log) {
 	c.Header("Content-Type", "text/plain")
 	if l.LogBytes != nil {
 		_, _ = c.Writer.Write(l.LogBytes)