feat: check pipelines' tokens before executing/creating pipeline (#8055)

backend/core/models/blueprint.go

@@ -92,3 +92,12 @@ type SyncPolicy struct {
 	TimeAfter  *time.Time `json:"timeAfter"`
 	TriggerSyncPolicy
 }
+
+type ConnectionTokenCheckResult struct {
+	PluginName   string `json:"pluginName" mapstructure:"pluginName"`
+	ConnectionID uint64 `json:"connectionId" mapstructure:"connectionId"`
+	Success      bool   `json:"success" mapstructure:"success"`
+	Message      string `json:"message" mapstructure:"message"`
+}
+
+type ApiBlueprintConnectionTokenCheck []ConnectionTokenCheckResult

backend/core/plugin/plugin_api.go

@@ -73,6 +73,10 @@ type PluginApi interface {
 	ApiResources() map[string]map[string]ApiResourceHandler
 }
 
+type PluginTestConnectionAPI interface {
+	TestConnection(id uint64) errors.Error
+}
+
 const wrapResponseError = "WRAP_RESPONSE_ERROR"
 
 func WrapTestConnectionErrResp(basicRes context.BasicRes, err errors.Error) errors.Error {

backend/core/plugin/plugin_blueprint.go

@@ -93,6 +93,16 @@ type ProjectMapper interface {
 	MapProject(projectName string, scopes []Scope) (models.PipelinePlan, errors.Error)
 }
 
+type ProjectTokenCheckerConnection struct {
+	PluginName   string
+	ConnectionId uint64
+}
+
+// ProjectTokenChecker is implemented by the plugin org, which generate a task tp check all connection's tokens
+type ProjectTokenChecker interface {
+	MakePipeline(skipCollectors bool, projectName string, scopes []ProjectTokenCheckerConnection) (models.PipelinePlan, errors.Error)
+}
+
 // CompositeDataSourcePluginBlueprintV200 is for unit test
 type CompositeDataSourcePluginBlueprintV200 interface {
 	PluginMeta

backend/helpers/pluginhelper/api/misc_helpers.go

@@ -18,9 +18,11 @@ limitations under the License.
 package api
 
 import (
+	"fmt"
 	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/errors"
 	"github.com/apache/incubator-devlake/core/models"
+	"github.com/apache/incubator-devlake/core/plugin"
 )
 
 // CallDB wraps DB calls with this signature, and handles the case if the struct is wrapped in a models.DynamicTabler.
@@ -31,3 +33,11 @@ func CallDB(f func(any, ...dal.Clause) errors.Error, x any, clauses ...dal.Claus
 	}
 	return f(x, clauses...)
 }
+
+func GenerateTestingConnectionApiResourceInput(connectionID uint64) *plugin.ApiResourceInput {
+	return &plugin.ApiResourceInput{
+		Params: map[string]string{
+			"connectionId": fmt.Sprintf("%d", connectionID),
+		},
+	}
+}

backend/plugins/ae/impl/impl.go

@@ -130,6 +130,11 @@ func (p AE) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p AE) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p AE) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/azuredevops_go/impl/impl.go

@@ -255,6 +255,11 @@ func (p Azuredevops) ApiResources() map[string]map[string]plugin.ApiResourceHand
 	}
 }
 
+func (p Azuredevops) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Azuredevops) MakeDataSourcePipelinePlanV200(
 	connectionId uint64,
 	scopes []*coreModels.BlueprintScope,

backend/plugins/bamboo/impl/impl.go

@@ -216,6 +216,11 @@ func (p Bamboo) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Bamboo) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Bamboo) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/bitbucket/impl/impl.go

@@ -211,6 +211,11 @@ func (p Bitbucket) MakeDataSourcePipelinePlanV200(
 	return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, skipCollectors)
 }
 
+func (p Bitbucket) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Bitbucket) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/bitbucket_server/impl/impl.go

@@ -170,6 +170,11 @@ func (p BitbucketServer) MakeDataSourcePipelinePlanV200(
 	return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, skipCollectors)
 }
 
+func (p BitbucketServer) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p BitbucketServer) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"connections/:connectionId/test": {

backend/plugins/circleci/impl/impl.go

@@ -176,6 +176,11 @@ func (p Circleci) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Circleci) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Circleci) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/customize/impl/impl.go

@@ -110,3 +110,7 @@ func (p Customize) ApiResources() map[string]map[string]plugin.ApiResourceHandle
 		},
 	}
 }
+
+func (p Customize) TestConnection(id uint64) errors.Error {
+	return nil
+}

backend/plugins/dbt/impl/impl.go

@@ -74,3 +74,7 @@ func (p Dbt) RootPkgPath() string {
 func (p Dbt) Name() string {
 	return "dbt"
 }
+
+func (p Dbt) TestConnection(id uint64) errors.Error {
+	return nil
+}

backend/plugins/dora/impl/impl.go

@@ -19,7 +19,6 @@ package impl
 
 import (
 	"encoding/json"
-
 	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/errors"
 	coreModels "github.com/apache/incubator-devlake/core/models"
@@ -168,3 +167,7 @@ func (p Dora) MakeMetricPluginPipelinePlanV200(projectName string, options json.
 	}
 	return plan, nil
 }
+
+func (p Dora) TestConnection(id uint64) errors.Error {
+	return nil
+}

backend/plugins/feishu/impl/impl.go

@@ -132,6 +132,11 @@ func (p Feishu) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Feishu) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Feishu) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/gitee/impl/impl.go

@@ -187,6 +187,11 @@ func (p Gitee) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Gitee) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Gitee) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/gitextractor/impl/impl.go

@@ -122,3 +122,7 @@ func (p GitExtractor) Close(taskCtx plugin.TaskContext) errors.Error {
 func (p GitExtractor) RootPkgPath() string {
 	return "github.com/apache/incubator-devlake/plugins/gitextractor"
 }
+
+func (p GitExtractor) TestConnection(id uint64) errors.Error {
+	return nil
+}

backend/plugins/github/api/shared.go

@@ -0,0 +1,42 @@
+/*
+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 api
+
+import (
+	"context"
+	"fmt"
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+)
+
+func TestExistingConnectionForTokenCheck(input *plugin.ApiResourceInput) errors.Error {
+	connection, err := dsHelper.ConnApi.GetMergedConnection(input)
+	if err != nil {
+		return err
+	}
+	testConnectionResult, testConnectionErr := testExistingConnection(context.TODO(), connection.GithubConn)
+	if testConnectionErr != nil {
+		return testConnectionErr
+	}
+	for _, token := range testConnectionResult.Tokens {
+		if !token.Success {
+			return errors.Default.New(fmt.Sprintf("token %s failed with msg: %s", token.Token, token.Message))
+		}
+	}
+	return nil
+}

backend/plugins/github/impl/impl.go

@@ -191,6 +191,10 @@ func (p Github) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Github) TestConnection(id uint64) errors.Error {
+	return api.TestExistingConnectionForTokenCheck(helper.GenerateTestingConnectionApiResourceInput(id))
+}
+
 func (p Github) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/gitlab/impl/impl.go

@@ -244,6 +244,11 @@ func (p Gitlab) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Gitlab) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Gitlab) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/icla/impl/impl.go

@@ -104,6 +104,10 @@ func (p Icla) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return nil
 }
 
+func (p Icla) TestConnection(id uint64) errors.Error {
+	return nil
+}
+
 func (p Icla) Close(taskCtx plugin.TaskContext) errors.Error {
 	data, ok := taskCtx.GetData().(*tasks.IclaTaskData)
 	if !ok {

backend/plugins/issue_trace/impl/enricher.go

@@ -19,7 +19,6 @@ package impl
 
 import (
 	"encoding/json"
-
 	"github.com/apache/incubator-devlake/core/context"
 	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/errors"
@@ -151,6 +150,10 @@ func (p IssueTrace) GetTablesInfo() []dal.Tabler {
 	}
 }
 
+func (p IssueTrace) TestConnection(id uint64) errors.Error {
+	return nil
+}
+
 func (p IssueTrace) MakeMetricPluginPipelinePlanV200(projectName string, options json.RawMessage) (coreModels.PipelinePlan, errors.Error) {
 	op := &tasks.Options{}
 	if options != nil && string(options) != "\"\"" {

backend/plugins/jenkins/impl/impl.go

@@ -177,6 +177,11 @@ func (p Jenkins) MakeDataSourcePipelinePlanV200(
 	return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, skipCollectors)
 }
 
+func (p Jenkins) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Jenkins) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/jira/impl/impl.go

@@ -285,6 +285,11 @@ func (p Jira) MigrationScripts() []plugin.MigrationScript {
 	return migrationscripts.All()
 }
 
+func (p Jira) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Jira) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/linker/impl/impl.go

@@ -125,3 +125,7 @@ func (p Linker) MakeMetricPluginPipelinePlanV200(projectName string, options jso
 	}
 	return plan, nil
 }
+
+func (p Linker) TestConnection(id uint64) errors.Error {
+	return nil
+}

backend/plugins/opsgenie/impl/impl.go

@@ -155,6 +155,11 @@ func (p Opsgenie) Close(taskCtx plugin.TaskContext) errors.Error {
 	return nil
 }
 
+func (p Opsgenie) TestConnection(id uint64) errors.Error {
+	_, err := api.TestExistingConnection(helper.GenerateTestingConnectionApiResourceInput(id))
+	return err
+}
+
 func (p Opsgenie) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
 	return map[string]map[string]plugin.ApiResourceHandler{
 		"test": {

backend/plugins/org/impl/impl.go

@@ -59,11 +59,33 @@ func (p Org) Name() string {
 
 func (p Org) SubTaskMetas() []plugin.SubTaskMeta {
 	return []plugin.SubTaskMeta{
+		tasks.TaskCheckTokenMeta,
 		tasks.ConnectUserAccountsExactMeta,
 		tasks.SetProjectMappingMeta,
 	}
 }
 
+func (p Org) MakePipeline(skipCollectors bool, projectName string, connections []plugin.ProjectTokenCheckerConnection) (coreModels.PipelinePlan, errors.Error) {
+	var plan coreModels.PipelinePlan
+	var stage coreModels.PipelineStage
+
+	// construct task options for Org
+	options := make(map[string]interface{})
+	if !skipCollectors {
+		options["projectConnections"] = connections
+	}
+
+	stage = append(stage, &coreModels.PipelineTask{
+		Plugin: p.Name(),
+		Subtasks: []string{
+			tasks.TaskCheckTokenMeta.Name,
+		},
+		Options: options,
+	})
+	plan = append(plan, stage)
+	return plan, nil
+}
+
 func (p Org) MapProject(projectName string, scopes []plugin.Scope) (coreModels.PipelinePlan, errors.Error) {
 	var plan coreModels.PipelinePlan
 	var stage coreModels.PipelineStage