feat(gitlab) fetch trigger jobs (apache#6167)

* feat(gitlab): fetching the "trigger" job as well

* feat(gitlab): add test for parsing trigger job

backend/plugins/gitlab/e2e/raw_tables/_raw_gitlab_api_trigger_job.csv

@@ -0,0 +1,2 @@
+id,params,data,url,input,created_at
+1,"{""ConnectionId"":1,""ProjectId"":44}","{""id"":1,""status"":""success"",""stage"":""deploy"",""name"":""deploy"",""ref"":""master"",""tag"":false,""coverage"":null,""allow_failure"":false,""created_at"":""2022-08-25T21:19:10.938Z"",""started_at"":null,""finished_at"":""2022-08-25T23:00:08.594Z"",""duration"":null,""user"":{""id"":39,""name"":"" chinese merico  chinese "",""username"":""merico"",""state"":""active"",""avatar_url"":""https://secure.gravatar.com/avatar/8d7684192b6b64e6a8e9a0b4127282a6?s=80\\u0026d=identicon"",""web_url"":""https://gitlab.nddtf.com/merico"",""created_at"":""2022-02-08T06:25:32.335Z"",""bio"":"""",""bio_html"":"""",""location"":"""",""public_email"":"""",""skype"":"""",""linkedin"":"""",""twitter"":"""",""website_url"":"""",""organization"":""merico"",""job_title"":"""",""work_information"":""merico""},""commit"":{""id"":""4ea55f395b3ca7195efd9b41132a5894ea277830"",""short_id"":""4ea55f39"",""created_at"":""2022-07-26T09:38:27.000+00:00"",""parent_ids"":[""fcca61fdfd70d9a79c3bb1449e199da86c1b34a7"",""5b8efd8aeb488925f45851e489a839ba975ca625""],""title"":""Merge branch 'test-mr' into 'master'"",""message"":""Merge branch 'test-mr' into 'master'\\n\\nUpdate  chinese .txt\\n\\nSee merge request merico/pip_test!1"",""author_name"":"" chinese merico  chinese "",""author_email"":""merico@nddtf.com"",""authored_date"":""2022-07-26T09:38:27.000+00:00"",""committer_name"":"" chinese merico  chinese "",""committer_email"":""merico@nddtf.com"",""committed_date"":""2022-07-26T09:38:27.000+00:00"",""web_url"":""https://gitlab.nddtf.com/merico/pip_test/-/commit/532f453a0942f78d5cf8e6e57ad75a3c1da83212""},""pipeline"":{""id"":73,""sha"":""532f453a0942f78d5cf8e6e57ad75a3c1da83212"",""ref"":""master"",""status"":""failed"",""created_at"":""2022-08-25T21:19:10.904Z"",""updated_at"":""2022-08-25T23:00:08.694Z"",""web_url"":""https://gitlab.nddtf.com/merico/pip_test/-/pipelines/73""},""web_url"":""https://gitlab.nddtf.com/merico/pip_test/-/jobs/152"",""artifacts"":[],""runner"":null,""artifacts_expire_at"":null}","https://gitlab.nddtf.com/api/v4/projects/44/jobs?page=1&per_page=100&sort=asc&with_stats=true",null,2022-08-26 13:55:39.057

backend/plugins/gitlab/e2e/snapshot_tables/_tool_gitlab_jobs_with_triggered.csv

@@ -0,0 +1,2 @@
+connection_id,gitlab_id,project_id,pipeline_id,status,stage,name,ref,tag,allow_failure,duration,web_url,gitlab_created_at,started_at,finished_at,_raw_data_params,_raw_data_table,_raw_data_id,_raw_data_remark
+1,1,44,73,success,deploy,deploy,master,0,0,0,"",2022-08-25T21:19:10.938+00:00,"",2022-08-25T23:00:08.594+00:00,"{""ConnectionId"":1,""ProjectId"":44}",_raw_gitlab_api_trigger_job,1,

backend/plugins/gitlab/e2e/trigger_job_test.go

@@ -0,0 +1,73 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e
+
+import (
+	"testing"
+
+	"github.com/apache/incubator-devlake/core/models/domainlayer/devops"
+	"github.com/apache/incubator-devlake/helpers/e2ehelper"
+	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+	"github.com/apache/incubator-devlake/plugins/gitlab/impl"
+	"github.com/apache/incubator-devlake/plugins/gitlab/models"
+	"github.com/apache/incubator-devlake/plugins/gitlab/tasks"
+)
+
+func TestGitlabTriggerJobDataFlow(t *testing.T) {
+
+	var gitlab impl.Gitlab
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "gitlab", gitlab)
+	regexEnricher := api.NewRegexEnricher()
+	_ = regexEnricher.TryAdd(devops.DEPLOYMENT, "(?i)compile")
+	taskData := &tasks.GitlabTaskData{
+		Options: &tasks.GitlabOptions{
+			ConnectionId: 1,
+			ProjectId:    44,
+		},
+		RegexEnricher: regexEnricher,
+	}
+	// import raw data table
+	// SELECT * FROM _raw_gitlab_api_job INTO OUTFILE "/tmp/_raw_gitlab_api_job.csv" FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"' LINES TERMINATED BY '\r\n';
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_gitlab_api_trigger_job.csv", "_raw_gitlab_api_trigger_job")
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_gitlab_api_job.csv", "_raw_gitlab_api_job")
+
+	// verify extraction
+	dataflowTester.FlushTabler(&models.GitlabJob{})
+	dataflowTester.Subtask(tasks.ExtractApiTriggerJobsMeta, taskData)
+	dataflowTester.VerifyTable(
+		models.GitlabJob{},
+		"./snapshot_tables/_tool_gitlab_jobs_with_triggered.csv",
+		e2ehelper.ColumnWithRawData(
+			"connection_id",
+			"gitlab_id",
+			"project_id",
+			"pipeline_id",
+			"status",
+			"stage",
+			"name",
+			"ref",
+			"tag",
+			"allow_failure",
+			"duration",
+			"web_url",
+			"gitlab_created_at",
+			"started_at",
+			"finished_at",
+		),
+	)
+}

backend/plugins/gitlab/tasks/trigger_job_collector.go

@@ -0,0 +1,92 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"github.com/apache/incubator-devlake/core/dal"
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+	helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+	"reflect"
+	"time"
+)
+
+func init() {
+	RegisterSubtaskMeta(&CollectApiTriggerJobsMeta)
+}
+
+const RAW_TRIGGER_JOB_TABLE = "gitlab_api_trigger_job"
+
+var CollectApiTriggerJobsMeta = plugin.SubTaskMeta{
+	Name:             "collectApiTriggerJobs",
+	EntryPoint:       CollectApiTriggerJobs,
+	EnabledByDefault: true,
+	Description:      "Collect job data from gitlab api, supports both timeFilter and diffSync.",
+	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
+	Dependencies:     []*plugin.SubTaskMeta{&ExtractApiPipelineDetailsMeta},
+}
+
+func CollectApiTriggerJobs(taskCtx plugin.SubTaskContext) errors.Error {
+	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_TRIGGER_JOB_TABLE)
+	collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+	tickInterval, err := helper.CalcTickInterval(200, 1*time.Minute)
+	incremental := collectorWithState.IsIncremental()
+
+	iterator, err := GetAllPipelinesIterator(taskCtx, collectorWithState)
+
+	err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
+		//RawDataSubTaskArgs: *rawDataSubTaskArgs,
+		ApiClient:       data.ApiClient,
+		MinTickInterval: &tickInterval,
+		PageSize:        100,
+		Incremental:     incremental,
+		Input:           iterator,
+		UrlTemplate:     "projects/{{ .Params.ProjectId }}/pipelines/{{ .Input.GitlabId }}/bridges",
+		ResponseParser:  GetRawMessageFromResponse,
+		AfterResponse:   ignoreHTTPStatus403, // ignore 403 for CI/CD disable
+	})
+
+	if err != nil {
+		return err
+	}
+
+	return collectorWithState.Execute()
+}
+
+func GetAllPipelinesIterator(taskCtx plugin.SubTaskContext, collectorWithState *helper.ApiCollectorStateManager) (*helper.DalCursorIterator, errors.Error) {
+	db := taskCtx.GetDal()
+	data := taskCtx.GetData().(*GitlabTaskData)
+	clauses := []dal.Clause{
+		dal.Select("gp.gitlab_id, gp.gitlab_id as iid"),
+		dal.From("_tool_gitlab_pipelines gp"),
+		dal.Where(
+			`gp.project_id = ? and gp.connection_id = ? and gp.gitlab_id not in (select json_extract(tj.input, '$.GitlabId') as gitlab_id from _raw_gitlab_api_trigger_job tj)`,
+			data.Options.ProjectId, data.Options.ConnectionId,
+		),
+	}
+	if collectorWithState.LatestState.LatestSuccessStart != nil {
+		clauses = append(clauses, dal.Where("gitlab_updated_at > ?", *collectorWithState.LatestState.LatestSuccessStart))
+	}
+	// construct the input iterator
+	cursor, err := db.Cursor(clauses...)
+	if err != nil {
+		return nil, err
+	}
+
+	return helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(GitlabInput{}))
+}

backend/plugins/gitlab/tasks/trigger_job_extractor.go

@@ -0,0 +1,106 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"encoding/json"
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+	"github.com/apache/incubator-devlake/plugins/gitlab/models"
+	"time"
+)
+
+func init() {
+	RegisterSubtaskMeta(&ExtractApiTriggerJobsMeta)
+}
+
+type ApiTriggerJob struct {
+	Id       int `json:"id"`
+	Status   string
+	Stage    string
+	Name     string
+	Ref      string
+	Duration float64
+	Pipeline struct {
+		Id int
+	}
+	CreatedAt  *time.Time `json:"created_at"`
+	StartedAt  *time.Time `json:"started_at"`
+	FinishedAt *time.Time `json:"finished_at"`
+}
+
+var ExtractApiTriggerJobsMeta = plugin.SubTaskMeta{
+	Name:             "extractApiTriggerJobs",
+	EntryPoint:       ExtractApiTriggerJobs,
+	EnabledByDefault: true,
+	Description:      "Extract raw Gitlab trigger jobs data into tool layer table GitlabPipeline",
+	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
+	Dependencies:     []*plugin.SubTaskMeta{&CollectApiTriggerJobsMeta},
+}
+
+func ExtractApiTriggerJobs(taskCtx plugin.SubTaskContext) errors.Error {
+	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_TRIGGER_JOB_TABLE)
+
+	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+		RawDataSubTaskArgs: *rawDataSubTaskArgs,
+		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+			// create gitlab commit
+			gitlabApiTriggerJob := &ApiTriggerJob{}
+			err := errors.Convert(json.Unmarshal(row.Data, gitlabApiTriggerJob))
+			if err != nil {
+				return nil, err
+			}
+
+			gitlabPipeline, err := convertTriggerJob(gitlabApiTriggerJob, data.Options.ProjectId)
+			if err != nil {
+				return nil, err
+			}
+
+			// use data.Options.ProjectId to set the value of ProjectId for it
+			gitlabPipeline.ProjectId = data.Options.ProjectId
+			gitlabPipeline.ConnectionId = data.Options.ConnectionId
+			results := make([]interface{}, 0, 1)
+			results = append(results, gitlabPipeline)
+
+			return results, nil
+		},
+	})
+
+	if err != nil {
+		return err
+	}
+
+	return extractor.Execute()
+}
+
+func convertTriggerJob(job *ApiTriggerJob, projectId int) (*models.GitlabJob, errors.Error) {
+	return &models.GitlabJob{
+		GitlabId:        job.Id,
+		ProjectId:       projectId,
+		Status:          job.Status,
+		Stage:           job.Stage,
+		Name:            job.Name,
+		Ref:             job.Ref,
+		Duration:        job.Duration,
+		PipelineId:      job.Pipeline.Id,
+		GitlabCreatedAt: job.CreatedAt,
+		StartedAt:       job.StartedAt,
+		FinishedAt:      job.FinishedAt,
+	}, nil
+}