You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@devlake.apache.org by ly...@apache.org on 2024/08/27 09:18:39 UTC
(incubator-devlake) 01/01: feat(github): doesn't init api client when re-transformating
This is an automated email from the ASF dual-hosted git repository.
lynwee pushed a commit to branch dev-1
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
commit 17d9b164aefce8f9e037b2cbf866e59e53bd5368
Author: d4x1 <15...@qq.com>
AuthorDate: Tue Aug 27 17:18:28 2024 +0800
feat(github): doesn't init api client when re-transformating
---
backend/core/runner/run_task.go | 2 +-
backend/plugins/github/impl/impl.go | 52 ++++++----
backend/plugins/github_graphql/impl/impl.go | 147 +++++++++++++++-------------
3 files changed, 109 insertions(+), 92 deletions(-)
diff --git a/backend/core/runner/run_task.go b/backend/core/runner/run_task.go
index fcc7d1101..0a12d7c4f 100644
--- a/backend/core/runner/run_task.go
+++ b/backend/core/runner/run_task.go
@@ -244,11 +244,11 @@ func RunPluginSubTasks(
defer closeablePlugin.Close(taskCtx)
}
options := task.Options
+ taskCtx.SetSyncPolicy(syncPolicy)
taskData, err := pluginTask.PrepareTaskData(taskCtx, options)
if err != nil {
return errors.Default.Wrap(err, fmt.Sprintf("error preparing task data for %s", task.Plugin))
}
- taskCtx.SetSyncPolicy(syncPolicy)
taskCtx.SetData(taskData)
// record subtasks sequence to DB
diff --git a/backend/plugins/github/impl/impl.go b/backend/plugins/github/impl/impl.go
index f73f33120..6b6cc12a8 100644
--- a/backend/plugins/github/impl/impl.go
+++ b/backend/plugins/github/impl/impl.go
@@ -137,11 +137,22 @@ func (p Github) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
if err != nil {
return nil, errors.Default.Wrap(err, "unable to get github connection by the given connection ID")
}
- apiClient, err := tasks.CreateApiClient(taskCtx, connection)
- if err != nil {
- return nil, errors.Default.Wrap(err, "unable to get github API client instance")
+
+ taskData := &tasks.GithubTaskData{
+ Options: op,
+ }
+
+ syncPolicy := taskCtx.SyncPolicy()
+ var apiClient *helper.ApiClient
+ if !syncPolicy.SkipCollectors {
+ newApiClient, err := tasks.CreateApiClient(taskCtx, connection)
+ if err != nil {
+ return nil, errors.Default.Wrap(err, "unable to get github API client instance")
+ }
+ apiClient = newApiClient.ApiClient
+ taskData.ApiClient = newApiClient
}
- err = EnrichOptions(taskCtx, op, apiClient.ApiClient)
+ err = EnrichOptions(taskCtx, op, apiClient)
if err != nil {
return nil, err
}
@@ -156,12 +167,7 @@ func (p Github) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
if err = regexEnricher.TryAdd(devops.ENV_NAME_PATTERN, op.ScopeConfig.EnvNamePattern); err != nil {
return nil, errors.BadInput.Wrap(err, "invalid value for `envNamePattern`")
}
-
- taskData := &tasks.GithubTaskData{
- Options: op,
- ApiClient: apiClient,
- RegexEnricher: regexEnricher,
- }
+ taskData.RegexEnricher = regexEnricher
return taskData, nil
}
@@ -239,7 +245,9 @@ func (p Github) Close(taskCtx plugin.TaskContext) errors.Error {
if !ok {
return errors.Default.New(fmt.Sprintf("GetData failed when try to close %+v", taskCtx))
}
- data.ApiClient.Release()
+ if data.ApiClient != nil {
+ data.ApiClient.Release()
+ }
return nil
}
@@ -266,17 +274,19 @@ func EnrichOptions(taskCtx plugin.TaskContext,
} else {
if taskCtx.GetDal().IsErrorNotFound(err) && op.Name != "" {
var repo *tasks.GithubApiRepo
- repo, err = api.MemorizedGetApiRepo(repo, op, apiClient)
- if err != nil {
- return err
- }
- logger.Debug(fmt.Sprintf("Current repo: %s", repo.FullName))
- scope := convertApiRepoToScope(repo, op.ConnectionId)
- err = taskCtx.GetDal().CreateIfNotExist(scope)
- if err != nil {
- return err
+ if apiClient != nil {
+ repo, err = api.MemorizedGetApiRepo(repo, op, apiClient)
+ if err != nil {
+ return err
+ }
+ logger.Debug(fmt.Sprintf("Current repo: %s", repo.FullName))
+ scope := convertApiRepoToScope(repo, op.ConnectionId)
+ err = taskCtx.GetDal().CreateIfNotExist(scope)
+ if err != nil {
+ return err
+ }
+ op.GithubId = repo.GithubId
}
- op.GithubId = repo.GithubId
} else {
return errors.Default.Wrap(err, fmt.Sprintf("fail to find repo %s", op.Name))
}
diff --git a/backend/plugins/github_graphql/impl/impl.go b/backend/plugins/github_graphql/impl/impl.go
index a1d77d176..d4953a18f 100644
--- a/backend/plugins/github_graphql/impl/impl.go
+++ b/backend/plugins/github_graphql/impl/impl.go
@@ -20,14 +20,13 @@ package impl
import (
"context"
"fmt"
+ "github.com/apache/incubator-devlake/core/models/domainlayer/devops"
"net/http"
"net/url"
"reflect"
"strings"
"time"
- "github.com/apache/incubator-devlake/core/models/domainlayer/devops"
-
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/log"
@@ -170,78 +169,88 @@ func (p GithubGraphql) PrepareTaskData(taskCtx plugin.TaskContext, options map[s
return nil, errors.Default.Wrap(err, "unable to get github connection by the given connection ID: %v")
}
- apiClient, err := githubTasks.CreateApiClient(taskCtx, connection)
- if err != nil {
- return nil, errors.Default.Wrap(err, "unable to get github API client instance")
- }
-
- err = githubImpl.EnrichOptions(taskCtx, &op, apiClient.ApiClient)
- if err != nil {
- return nil, err
+ taskData := &githubTasks.GithubTaskData{
+ Options: &op,
}
- tokens := strings.Split(connection.Token, ",")
- src := oauth2.StaticTokenSource(
- &oauth2.Token{AccessToken: tokens[0]},
- )
- oauthContext := taskCtx.GetContext()
- proxy := connection.GetProxy()
- if proxy != "" {
- pu, err := url.Parse(proxy)
+ syncPolicy := taskCtx.SyncPolicy()
+ var enricherApiClient *helper.ApiClient
+ if !syncPolicy.SkipCollectors {
+ apiClient, err := githubTasks.CreateApiClient(taskCtx, connection)
if err != nil {
- return nil, errors.Convert(err)
+ return nil, errors.Default.Wrap(err, "unable to get github API client instance")
}
- if pu.Scheme == "http" || pu.Scheme == "socks5" {
- proxyClient := &http.Client{
- Transport: &http.Transport{Proxy: http.ProxyURL(pu)},
+ enricherApiClient = apiClient.ApiClient
+
+ tokens := strings.Split(connection.Token, ",")
+ src := oauth2.StaticTokenSource(
+ &oauth2.Token{AccessToken: tokens[0]},
+ )
+ oauthContext := taskCtx.GetContext()
+ proxy := connection.GetProxy()
+ if proxy != "" {
+ pu, err := url.Parse(proxy)
+ if err != nil {
+ return nil, errors.Convert(err)
+ }
+ if pu.Scheme == "http" || pu.Scheme == "socks5" {
+ proxyClient := &http.Client{
+ Transport: &http.Transport{Proxy: http.ProxyURL(pu)},
+ }
+ oauthContext = context.WithValue(
+ taskCtx.GetContext(),
+ oauth2.HTTPClient,
+ proxyClient,
+ )
+ logger.Debug("Proxy set in oauthContext to %s", proxy)
+ } else {
+ return nil, errors.BadInput.New("Unsupported scheme set in proxy")
}
- oauthContext = context.WithValue(
- taskCtx.GetContext(),
- oauth2.HTTPClient,
- proxyClient,
- )
- logger.Debug("Proxy set in oauthContext to %s", proxy)
- } else {
- return nil, errors.BadInput.New("Unsupported scheme set in proxy")
}
- }
- httpClient := oauth2.NewClient(oauthContext, src)
- endpoint, err := errors.Convert01(url.Parse(connection.Endpoint))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, fmt.Sprintf("malformed connection endpoint supplied: %s", connection.Endpoint))
- }
+ httpClient := oauth2.NewClient(oauthContext, src)
+ endpoint, err := errors.Convert01(url.Parse(connection.Endpoint))
+ if err != nil {
+ return nil, errors.BadInput.Wrap(err, fmt.Sprintf("malformed connection endpoint supplied: %s", connection.Endpoint))
+ }
- // github.com and github enterprise have different graphql endpoints
- endpoint.Path = "/graphql" // see https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
- if endpoint.Hostname() != "api.github.com" {
- // see https://docs.github.com/en/enterprise-server@3.11/graphql/guides/forming-calls-with-graphql
- endpoint.Path = "/api/graphql"
- }
- client := graphql.NewClient(endpoint.String(), httpClient)
- graphqlClient, err := helper.CreateAsyncGraphqlClient(taskCtx, client, taskCtx.GetLogger(),
- func(ctx context.Context, client *graphql.Client, logger log.Logger) (rateRemaining int, resetAt *time.Time, err errors.Error) {
- var query GraphQueryRateLimit
- dataErrors, err := errors.Convert01(client.Query(taskCtx.GetContext(), &query, nil))
- if err != nil {
- return 0, nil, err
- }
- if len(dataErrors) > 0 {
- return 0, nil, errors.Default.Wrap(dataErrors[0], `query rate limit fail`)
- }
- logger.Info(`github graphql init success with remaining %d/%d and will reset at %s`,
- query.RateLimit.Remaining, query.RateLimit.Limit, query.RateLimit.ResetAt)
- return int(query.RateLimit.Remaining), &query.RateLimit.ResetAt, nil
+ // github.com and github enterprise have different graphql endpoints
+ endpoint.Path = "/graphql" // see https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
+ if endpoint.Hostname() != "api.github.com" {
+ // see https://docs.github.com/en/enterprise-server@3.11/graphql/guides/forming-calls-with-graphql
+ endpoint.Path = "/api/graphql"
+ }
+ client := graphql.NewClient(endpoint.String(), httpClient)
+ graphqlClient, err := helper.CreateAsyncGraphqlClient(taskCtx, client, taskCtx.GetLogger(),
+ func(ctx context.Context, client *graphql.Client, logger log.Logger) (rateRemaining int, resetAt *time.Time, err errors.Error) {
+ var query GraphQueryRateLimit
+ dataErrors, err := errors.Convert01(client.Query(taskCtx.GetContext(), &query, nil))
+ if err != nil {
+ return 0, nil, err
+ }
+ if len(dataErrors) > 0 {
+ return 0, nil, errors.Default.Wrap(dataErrors[0], `query rate limit fail`)
+ }
+ logger.Info(`github graphql init success with remaining %d/%d and will reset at %s`,
+ query.RateLimit.Remaining, query.RateLimit.Limit, query.RateLimit.ResetAt)
+ return int(query.RateLimit.Remaining), &query.RateLimit.ResetAt, nil
+ })
+ if err != nil {
+ return nil, err
+ }
+
+ graphqlClient.SetGetRateCost(func(q interface{}) int {
+ v := reflect.ValueOf(q)
+ return int(v.Elem().FieldByName(`RateLimit`).FieldByName(`Cost`).Int())
})
+ taskData.ApiClient = apiClient
+ taskData.GraphqlClient = graphqlClient
+ }
+ err = githubImpl.EnrichOptions(taskCtx, &op, enricherApiClient)
if err != nil {
return nil, err
}
- graphqlClient.SetGetRateCost(func(q interface{}) int {
- v := reflect.ValueOf(q)
- return int(v.Elem().FieldByName(`RateLimit`).FieldByName(`Cost`).Int())
- })
-
regexEnricher := helper.NewRegexEnricher()
if err = regexEnricher.TryAdd(devops.DEPLOYMENT, op.ScopeConfig.DeploymentPattern); err != nil {
return nil, errors.BadInput.Wrap(err, "invalid value for `deploymentPattern`")
@@ -252,13 +261,7 @@ func (p GithubGraphql) PrepareTaskData(taskCtx plugin.TaskContext, options map[s
if err = regexEnricher.TryAdd(devops.ENV_NAME_PATTERN, op.ScopeConfig.EnvNamePattern); err != nil {
return nil, errors.BadInput.Wrap(err, "invalid value for `envNamePattern`")
}
-
- taskData := &githubTasks.GithubTaskData{
- Options: &op,
- ApiClient: apiClient,
- GraphqlClient: graphqlClient,
- RegexEnricher: regexEnricher,
- }
+ taskData.RegexEnricher = regexEnricher
return taskData, nil
}
@@ -277,8 +280,12 @@ func (p GithubGraphql) Close(taskCtx plugin.TaskContext) errors.Error {
if !ok {
return errors.Default.New(fmt.Sprintf("GetData failed when try to close %+v", taskCtx))
}
- data.ApiClient.Release()
- data.GraphqlClient.Release()
+ if data.ApiClient != nil {
+ data.ApiClient.Release()
+ }
+ if data.GraphqlClient != nil {
+ data.GraphqlClient.Release()
+ }
return nil
}