feat(gitlab): collect child pipelines

backend/plugins/gitlab/models/migrationscripts/20240906_add_source_to_pipelines.go

@@ -0,0 +1,60 @@
+/*
+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 migrationscripts
+
+import (
+	"github.com/apache/incubator-devlake/core/context"
+	"github.com/apache/incubator-devlake/core/errors"
+)
+
+type gitlabPipelineProject240906 struct {
+	Source string
+}
+
+func (gitlabPipelineProject240906) TableName() string {
+	return "_tool_gitlab_pipeline_projects"
+}
+
+type gitlabPipeline240906 struct {
+	Source string
+}
+
+func (gitlabPipeline240906) TableName() string {
+	return "_tool_gitlab_pipelines"
+}
+
+type addIsChildToPipelines240906 struct{}
+
+func (*addIsChildToPipelines240906) Up(baseRes context.BasicRes) errors.Error {
+	db := baseRes.GetDal()
+	if err := db.AutoMigrate(&gitlabPipeline240906{}); err != nil {
+		return err
+	}
+	if err := db.AutoMigrate(&gitlabPipelineProject240906{}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (*addIsChildToPipelines240906) Version() uint64 {
+	return 20240906150000
+}
+
+func (*addIsChildToPipelines240906) Name() string {
+	return "add is_child to table _tool_gitlab_pipelines and _tool_gitlab_pipeline_projects"
+}

backend/plugins/gitlab/models/migrationscripts/register.go

@@ -51,5 +51,6 @@ func All() []plugin.MigrationScript {
 		new(addGitlabAssignee),
 		new(addGitlabAssigneeAndReviewerPrimaryKey),
 		new(changeIssueComponentType),
+		new(addIsChildToPipelines240906),
 	}
 }

backend/plugins/gitlab/models/pipeline.go

@@ -23,6 +23,10 @@ import (
 	"github.com/apache/incubator-devlake/core/models/common"
 )
 
+const (
+	PipelineSourceParentPipeline = "parent_pipeline"
+)
+
 type GitlabPipeline struct {
 	ConnectionId uint64 `gorm:"primaryKey"`
 
@@ -45,10 +49,16 @@ type GitlabPipeline struct {
 	Environment string `gorm:"type:varchar(255)"`
 
 	IsDetailRequired bool
+	Source           string
 
 	common.NoPKModel
 }
 
+func (gitlabPipeline GitlabPipeline) GenerateIsChild() bool {
+	return gitlabPipeline.Source == PipelineSourceParentPipeline
+
+}
+
 func (GitlabPipeline) TableName() string {
 	return "_tool_gitlab_pipelines"
 }
@@ -63,6 +73,8 @@ type GitlabPipelineProject struct {
 	GitlabCreatedAt *time.Time
 	GitlabUpdatedAt *time.Time
 	common.NoPKModel
+
+	Source string
 }
 
 func (GitlabPipelineProject) TableName() string {

backend/plugins/gitlab/tasks/child_pipeline_collector.go

@@ -0,0 +1,82 @@
+/*
+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 (
+	"fmt"
+	"net/url"
+	"time"
+
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+	helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+)
+
+func init() {
+	RegisterSubtaskMeta(&CollectApiChildPipelinesMeta)
+}
+
+const RAW_CHILD_PIPELINE_TABLE = "gitlab_api_child_pipeline"
+
+var CollectApiChildPipelinesMeta = plugin.SubTaskMeta{
+	Name:             "Collect Child Pipelines",
+	EntryPoint:       CollectApiChildPipelines,
+	EnabledByDefault: true,
+	Description:      "Collect child pipeline data from gitlab api, supports both timeFilter and diffSync.",
+	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
+}
+
+func CollectApiChildPipelines(taskCtx plugin.SubTaskContext) errors.Error {
+	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_CHILD_PIPELINE_TABLE)
+	apiCollector, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
+	if err != nil {
+		return err
+	}
+
+	tickInterval, err := helper.CalcTickInterval(200, 1*time.Minute)
+	if err != nil {
+		return err
+	}
+
+	err = apiCollector.InitCollector(helper.ApiCollectorArgs{
+		RawDataSubTaskArgs: *rawDataSubTaskArgs,
+		ApiClient:          data.ApiClient,
+		MinTickInterval:    &tickInterval,
+		PageSize:           100,
+		UrlTemplate:        "projects/{{ .Params.ProjectId }}/pipelines",
+		Query: func(reqData *helper.RequestData) (url.Values, errors.Error) {
+			query := url.Values{}
+			if apiCollector.GetSince() != nil {
+				query.Set("updated_after", apiCollector.GetSince().Format(time.RFC3339))
+			}
+			query.Set("with_stats", "true")
+			query.Set("sort", "asc")
+			query.Set("source", "parent_pipeline")
+			query.Set("page", fmt.Sprintf("%v", reqData.Pager.Page))
+			query.Set("per_page", fmt.Sprintf("%v", reqData.Pager.Size))
+			return query, nil
+		},
+		ResponseParser: GetRawMessageFromResponse,
+		AfterResponse:  ignoreHTTPStatus403, // ignore 403 for CI/CD disable
+	})
+	if err != nil {
+		return err
+	}
+
+	return apiCollector.Execute()
+}

backend/plugins/gitlab/tasks/child_pipeline_extractor.go

@@ -0,0 +1,84 @@
+/*
+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/models/common"
+	"github.com/apache/incubator-devlake/core/plugin"
+	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+	"github.com/apache/incubator-devlake/plugins/gitlab/models"
+)
+
+func init() {
+	RegisterSubtaskMeta(&ExtractApiChildPipelinesMeta)
+}
+
+var ExtractApiChildPipelinesMeta = plugin.SubTaskMeta{
+	Name:             "Extract Child Pipelines",
+	EntryPoint:       ExtractApiChildPipelines,
+	EnabledByDefault: true,
+	Description:      "Extract raw pipelines data into tool layer table GitlabPipeline",
+	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
+	Dependencies:     []*plugin.SubTaskMeta{&CollectApiChildPipelinesMeta},
+}
+
+func ExtractApiChildPipelines(taskCtx plugin.SubTaskContext) errors.Error {
+	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_CHILD_PIPELINE_TABLE)
+
+	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+		RawDataSubTaskArgs: *rawDataSubTaskArgs,
+		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+			gitlabApiChildPipeline := &ApiPipeline{}
+			err := errors.Convert(json.Unmarshal(row.Data, gitlabApiChildPipeline))
+			if err != nil {
+				return nil, err
+			}
+			pipelineProject := convertApiPipelineToGitlabPipelineProject(gitlabApiChildPipeline, data.Options.ConnectionId, data.Options.ProjectId)
+			return []interface{}{pipelineProject}, nil
+		},
+	})
+
+	if err != nil {
+		return err
+	}
+
+	err = extractor.Execute()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func convertApiPipelineToGitlabPipelineProject(gitlabApiChildPipeline *ApiPipeline, connectionId uint64, projectId int) *models.GitlabPipelineProject {
+	pipelineProject := &models.GitlabPipelineProject{
+		ConnectionId:    connectionId,
+		PipelineId:      gitlabApiChildPipeline.Id,
+		ProjectId:       projectId,
+		Ref:             gitlabApiChildPipeline.Ref,
+		WebUrl:          gitlabApiChildPipeline.WebUrl,
+		Sha:             gitlabApiChildPipeline.Sha,
+		Source:          gitlabApiChildPipeline.Source,
+		GitlabCreatedAt: common.Iso8601TimeToTime(gitlabApiChildPipeline.CreatedAt),
+		GitlabUpdatedAt: common.Iso8601TimeToTime(gitlabApiChildPipeline.UpdatedAt),
+	}
+	return pipelineProject
+}

backend/plugins/gitlab/tasks/job_extractor.go

@@ -65,7 +65,6 @@ func ExtractApiJobs(taskCtx plugin.SubTaskContext) errors.Error {
 	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
 		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
-			// create gitlab commit
 			gitlabApiJob := &ApiJob{}
 			err := errors.Convert(json.Unmarshal(row.Data, gitlabApiJob))
 			if err != nil {

backend/plugins/gitlab/tasks/pipeline_detail_collector.go

@@ -40,7 +40,7 @@ var CollectApiPipelineDetailsMeta = plugin.SubTaskMeta{
 	EnabledByDefault: true,
 	Description:      "Collect pipeline details data from gitlab api, supports both timeFilter and diffSync.",
 	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
-	Dependencies:     []*plugin.SubTaskMeta{&ExtractApiPipelinesMeta},
+	Dependencies:     []*plugin.SubTaskMeta{&ExtractApiPipelinesMeta, &ExtractApiChildPipelinesMeta},
 }
 
 func CollectApiPipelineDetails(taskCtx plugin.SubTaskContext) errors.Error {

backend/plugins/gitlab/tasks/pipeline_detail_convertor.go

@@ -104,7 +104,8 @@ func ConvertDetailPipelines(taskCtx plugin.SubTaskContext) errors.Error {
 				Type:              gitlabPipeline.Type,
 				DurationSec:       float64(gitlabPipeline.Duration),
 				// DisplayTitle:      gitlabPipeline.Ref,
-				Url: gitlabPipeline.WebUrl,
+				Url:     gitlabPipeline.WebUrl,
+				IsChild: gitlabPipeline.GenerateIsChild(),
 			}
 			return []interface{}{
 				domainPipeline,

backend/plugins/gitlab/tasks/pipeline_detail_extractor.go

@@ -47,7 +47,6 @@ func ExtractApiPipelineDetails(taskCtx plugin.SubTaskContext) errors.Error {
 	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
 		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
-			// create gitlab commit
 			gitlabApiPipeline := &ApiPipeline{}
 			err := errors.Convert(json.Unmarshal(row.Data, gitlabApiPipeline))
 			if err != nil {
@@ -70,11 +69,10 @@ func ExtractApiPipelineDetails(taskCtx plugin.SubTaskContext) errors.Error {
 				ConnectionId:    data.Options.ConnectionId,
 				Type:            data.RegexEnricher.ReturnNameIfMatched(devops.DEPLOYMENT, gitlabApiPipeline.Ref),
 				Environment:     data.RegexEnricher.ReturnNameIfMatched(devops.PRODUCTION, gitlabApiPipeline.Ref),
+				Source:          gitlabApiPipeline.Source,
 			}
 
-			results := make([]interface{}, 0, 1)
-			results = append(results, gitlabPipeline)
-			return results, nil
+			return []interface{}{gitlabPipeline}, nil
 		},
 	})
 	if err != nil {

backend/plugins/gitlab/tasks/pipeline_extractor.go

@@ -24,7 +24,6 @@ import (
 	"github.com/apache/incubator-devlake/core/models/common"
 	"github.com/apache/incubator-devlake/core/plugin"
 	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
-	"github.com/apache/incubator-devlake/plugins/gitlab/models"
 )
 
 func init() {
@@ -51,6 +50,7 @@ type ApiPipeline struct {
 	Duration       int
 	QueuedDuration *float64 `json:"queued_duration"`
 	WebUrl         string   `json:"web_url"`
+	Source         string   `json:"source"`
 
 	CreatedAt  *common.Iso8601Time `json:"created_at"`
 	UpdatedAt  *common.Iso8601Time `json:"updated_at"`
@@ -75,28 +75,13 @@ func ExtractApiPipelines(taskCtx plugin.SubTaskContext) errors.Error {
 	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
 		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
-			// create gitlab commit
 			gitlabApiPipeline := &ApiPipeline{}
 			err := errors.Convert(json.Unmarshal(row.Data, gitlabApiPipeline))
 			if err != nil {
 				return nil, err
 			}
-
-			pipelineProject := &models.GitlabPipelineProject{
-				ConnectionId:    data.Options.ConnectionId,
-				PipelineId:      gitlabApiPipeline.Id,
-				ProjectId:       data.Options.ProjectId,
-				Ref:             gitlabApiPipeline.Ref,
-				WebUrl:          gitlabApiPipeline.WebUrl,
-				Sha:             gitlabApiPipeline.Sha,
-				GitlabCreatedAt: common.Iso8601TimeToTime(gitlabApiPipeline.CreatedAt),
-				GitlabUpdatedAt: common.Iso8601TimeToTime(gitlabApiPipeline.UpdatedAt),
-			}
-
-			results := make([]interface{}, 0, 1)
-			results = append(results, pipelineProject)
-
-			return results, nil
+			pipelineProject := convertApiPipelineToGitlabPipelineProject(gitlabApiPipeline, data.Options.ConnectionId, data.Options.ProjectId)
+			return []interface{}{pipelineProject}, nil
 		},
 	})
 

backend/plugins/gitlab/tasks/tag_extractor.go

@@ -56,10 +56,9 @@ func ExtractApiTag(taskCtx plugin.SubTaskContext) errors.Error {
 	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
 		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
-			// need to extract 1 kinds of entities here
+			// need to extract 1 kind of entities here
 			results := make([]interface{}, 0, 1)
 
-			// create gitlab commit
 			gitlabApiTag := &GitlabApiTag{}
 			err := errors.Convert(json.Unmarshal(row.Data, gitlabApiTag))
 			if err != nil {

backend/plugins/gitlab/tasks/trigger_job_extractor.go

@@ -61,7 +61,6 @@ func ExtractApiTriggerJobs(taskCtx plugin.SubTaskContext) errors.Error {
 	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 {