You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@devlake.apache.org by ab...@apache.org on 2023/12/05 09:58:40 UTC
(incubator-devlake) branch github-graphql-split updated: fix: split the collector into collector+extractor
This is an automated email from the ASF dual-hosted git repository.
abeizn pushed a commit to branch github-graphql-split
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
The following commit(s) were added to refs/heads/github-graphql-split by this push:
new 16e5fe863 fix: split the collector into collector+extractor
16e5fe863 is described below
commit 16e5fe8635e0741dd6f3fe915ac8edb967150998
Author: abeizn <zi...@merico.dev>
AuthorDate: Tue Dec 5 17:58:29 2023 +0800
fix: split the collector into collector+extractor
---
backend/plugins/github_graphql/impl/impl.go | 75 ++++-----
.../github_graphql/tasks/account_collector.go | 41 +----
.../github_graphql/tasks/account_extractor.go | 106 ++++++++++++
...lector_extractor.go => deployment_collector.go} | 64 +-------
.../github_graphql/tasks/deployment_extractor.go | 104 ++++++++++++
.../github_graphql/tasks/issue_collector.go | 160 +-----------------
.../{issue_collector.go => issue_extractor.go} | 124 +++-----------
.../plugins/github_graphql/tasks/job_collector.go | 52 +-----
.../plugins/github_graphql/tasks/job_extractor.go | 106 ++++++++++++
.../plugins/github_graphql/tasks/pr_collector.go | 154 +-----------------
.../tasks/{pr_collector.go => pr_extractor.go} | 180 ++++-----------------
11 files changed, 434 insertions(+), 732 deletions(-)
diff --git a/backend/plugins/github_graphql/impl/impl.go b/backend/plugins/github_graphql/impl/impl.go
index bf1808663..d132c800b 100644
--- a/backend/plugins/github_graphql/impl/impl.go
+++ b/backend/plugins/github_graphql/impl/impl.go
@@ -81,53 +81,56 @@ func (p GithubGraphql) GetTablesInfo() []dal.Tabler {
func (p GithubGraphql) SubTaskMetas() []plugin.SubTaskMeta {
return []plugin.SubTaskMeta{
- //tasks.CollectRepoMeta,
-
// collect millstones
githubTasks.CollectMilestonesMeta,
githubTasks.ExtractMilestonesMeta,
// collect issue & pr, deps on millstone
- tasks.CollectIssueMeta,
- tasks.CollectPrMeta,
+ tasks.CollectIssuesMeta,
+ tasks.ExtractIssuesMeta,
+ tasks.CollectPrsMeta,
+ tasks.ExtractPrsMeta,
// collect workflow run & job
githubTasks.CollectRunsMeta,
githubTasks.ExtractRunsMeta,
- tasks.CollectGraphqlJobsMeta,
-
- // collect others
- githubTasks.CollectApiCommentsMeta,
- githubTasks.ExtractApiCommentsMeta,
- githubTasks.CollectApiEventsMeta,
- githubTasks.ExtractApiEventsMeta,
- githubTasks.CollectApiPrReviewCommentsMeta,
- githubTasks.ExtractApiPrReviewCommentsMeta,
-
- // collect account, deps on all before
- tasks.CollectAccountMeta,
-
- // convert to domain layer
- githubTasks.ConvertRunsMeta,
- githubTasks.ConvertJobsMeta,
- githubTasks.EnrichPullRequestIssuesMeta,
- githubTasks.ConvertRepoMeta,
- githubTasks.ConvertIssuesMeta,
- githubTasks.ConvertCommitsMeta,
- githubTasks.ConvertIssueLabelsMeta,
- githubTasks.ConvertPullRequestCommitsMeta,
- githubTasks.ConvertPullRequestsMeta,
- githubTasks.ConvertPullRequestReviewsMeta,
- githubTasks.ConvertPullRequestLabelsMeta,
- githubTasks.ConvertPullRequestIssuesMeta,
- githubTasks.ConvertIssueAssigneeMeta,
- githubTasks.ConvertIssueCommentsMeta,
- githubTasks.ConvertPullRequestCommentsMeta,
- githubTasks.ConvertMilestonesMeta,
- githubTasks.ConvertAccountsMeta,
+ tasks.CollectJobsMeta,
+ tasks.ExtractJobsMeta,
+
+ // // collect others
+ // githubTasks.CollectApiCommentsMeta,
+ // githubTasks.ExtractApiCommentsMeta,
+ // githubTasks.CollectApiEventsMeta,
+ // githubTasks.ExtractApiEventsMeta,
+ // githubTasks.CollectApiPrReviewCommentsMeta,
+ // githubTasks.ExtractApiPrReviewCommentsMeta,
+
+ // // collect account, deps on all before
+ // tasks.CollectAccountMeta,
+ // tasks.ExtractAccountsMeta,
+
+ // // convert to domain layer
+ // githubTasks.ConvertRunsMeta,
+ // githubTasks.ConvertJobsMeta,
+ // githubTasks.EnrichPullRequestIssuesMeta,
+ // githubTasks.ConvertRepoMeta,
+ // githubTasks.ConvertIssuesMeta,
+ // githubTasks.ConvertCommitsMeta,
+ // githubTasks.ConvertIssueLabelsMeta,
+ // githubTasks.ConvertPullRequestCommitsMeta,
+ // githubTasks.ConvertPullRequestsMeta,
+ // githubTasks.ConvertPullRequestReviewsMeta,
+ // githubTasks.ConvertPullRequestLabelsMeta,
+ // githubTasks.ConvertPullRequestIssuesMeta,
+ // githubTasks.ConvertIssueAssigneeMeta,
+ // githubTasks.ConvertIssueCommentsMeta,
+ // githubTasks.ConvertPullRequestCommentsMeta,
+ // githubTasks.ConvertMilestonesMeta,
+ // githubTasks.ConvertAccountsMeta,
// deployment
- tasks.CollectAndExtractDeploymentsMeta,
+ tasks.CollectDeploymentsMeta,
+ tasks.ExtractDeploymentsMeta,
githubTasks.ConvertDeploymentsMeta,
}
}
diff --git a/backend/plugins/github_graphql/tasks/account_collector.go b/backend/plugins/github_graphql/tasks/account_collector.go
index 73d627239..3aed8c6a9 100644
--- a/backend/plugins/github_graphql/tasks/account_collector.go
+++ b/backend/plugins/github_graphql/tasks/account_collector.go
@@ -123,18 +123,7 @@ func CollectAccount(taskCtx plugin.SubTaskContext) errors.Error {
// log and ignore
taskCtx.GetLogger().Warn(dataError, `query user get error but ignore`)
}
- query := iQuery.(*GraphqlQueryAccountWrapper)
- accounts := query.Users
-
- results := make([]interface{}, 0, 1)
- for _, account := range accounts {
- relatedUsers, err := convertAccount(account, data.Options.ConnectionId)
- if err != nil {
- return nil, err
- }
- results = append(results, relatedUsers...)
- }
- return results, nil
+ return nil, nil
},
})
@@ -144,31 +133,3 @@ func CollectAccount(taskCtx plugin.SubTaskContext) errors.Error {
return collector.Execute()
}
-
-func convertAccount(res GraphqlQueryAccount, connId uint64) ([]interface{}, errors.Error) {
- results := make([]interface{}, 0, len(res.Organizations.Nodes)+1)
- githubAccount := &models.GithubAccount{
- ConnectionId: connId,
- Id: res.Id,
- Login: res.Login,
- Name: res.Name,
- Company: res.Company,
- Email: res.Email,
- AvatarUrl: res.AvatarUrl,
- //Url: res.Url,
- HtmlUrl: res.HtmlUrl,
- Type: `User`,
- }
- results = append(results, githubAccount)
- for _, apiAccountOrg := range res.Organizations.Nodes {
- githubAccountOrg := &models.GithubAccountOrg{
- ConnectionId: connId,
- AccountId: res.Id,
- OrgId: apiAccountOrg.DatabaseId,
- OrgLogin: apiAccountOrg.Login,
- }
- results = append(results, githubAccountOrg)
- }
-
- return results, nil
-}
diff --git a/backend/plugins/github_graphql/tasks/account_extractor.go b/backend/plugins/github_graphql/tasks/account_extractor.go
new file mode 100644
index 000000000..6c586bbba
--- /dev/null
+++ b/backend/plugins/github_graphql/tasks/account_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/github/models"
+ githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
+)
+
+var _ plugin.SubTaskEntryPoint = ExtractAccounts
+
+var ExtractAccountsMeta = plugin.SubTaskMeta{
+ Name: "extractAccounts",
+ EntryPoint: ExtractAccounts,
+ EnabledByDefault: true,
+ Description: "extract raw account data into tool layer table github_accounts",
+ DomainTypes: []string{plugin.DOMAIN_TYPE_CROSS},
+}
+
+func ExtractAccounts(taskCtx plugin.SubTaskContext) errors.Error {
+ data := taskCtx.GetData().(*githubTasks.GithubTaskData)
+ extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+ RawDataSubTaskArgs: api.RawDataSubTaskArgs{
+ Ctx: taskCtx,
+ Params: githubTasks.GithubApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ Name: data.Options.Name,
+ },
+ Table: RAW_ACCOUNTS_TABLE,
+ },
+ Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+ apiAccount := &GraphqlQueryAccountWrapper{}
+ err := errors.Convert(json.Unmarshal(row.Data, apiAccount))
+ if err != nil {
+ return nil, err
+ }
+ accounts := apiAccount.Users
+ results := make([]interface{}, 0, 1)
+ for _, account := range accounts {
+ relatedUsers, err := convertAccount(account, data.Options.ConnectionId)
+ if err != nil {
+ return nil, err
+ }
+ results = append(results, relatedUsers...)
+ }
+ return results, nil
+ },
+ })
+
+ if err != nil {
+ return err
+ }
+
+ return extractor.Execute()
+}
+
+func convertAccount(res GraphqlQueryAccount, connId uint64) ([]interface{}, errors.Error) {
+ results := make([]interface{}, 0, len(res.Organizations.Nodes)+1)
+ githubAccount := &models.GithubAccount{
+ ConnectionId: connId,
+ Id: res.Id,
+ Login: res.Login,
+ Name: res.Name,
+ Company: res.Company,
+ Email: res.Email,
+ AvatarUrl: res.AvatarUrl,
+ //Url: res.Url,
+ HtmlUrl: res.HtmlUrl,
+ Type: `User`,
+ }
+ if githubAccount.Id == 0 {
+ return nil, nil
+ }
+ results = append(results, githubAccount)
+ for _, apiAccountOrg := range res.Organizations.Nodes {
+ githubAccountOrg := &models.GithubAccountOrg{
+ ConnectionId: connId,
+ AccountId: res.Id,
+ OrgId: apiAccountOrg.DatabaseId,
+ OrgLogin: apiAccountOrg.Login,
+ }
+ results = append(results, githubAccountOrg)
+ }
+
+ return results, nil
+}
diff --git a/backend/plugins/github_graphql/tasks/deployment_collector_extractor.go b/backend/plugins/github_graphql/tasks/deployment_collector.go
similarity index 65%
rename from backend/plugins/github_graphql/tasks/deployment_collector_extractor.go
rename to backend/plugins/github_graphql/tasks/deployment_collector.go
index ced50bf20..c6e686fd8 100644
--- a/backend/plugins/github_graphql/tasks/deployment_collector_extractor.go
+++ b/backend/plugins/github_graphql/tasks/deployment_collector.go
@@ -22,23 +22,21 @@ import (
"time"
"github.com/apache/incubator-devlake/core/errors"
- "github.com/apache/incubator-devlake/core/models/common"
"github.com/apache/incubator-devlake/core/plugin"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
- githubModels "github.com/apache/incubator-devlake/plugins/github/models"
githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
"github.com/merico-dev/graphql"
)
-var _ plugin.SubTaskEntryPoint = CollectAndExtractDeployments
+var _ plugin.SubTaskEntryPoint = CollectDeployments
const (
RAW_DEPLOYMENT = "github_graphql_deployment"
)
-var CollectAndExtractDeploymentsMeta = plugin.SubTaskMeta{
- Name: "CollectAndExtractDeployments",
- EntryPoint: CollectAndExtractDeployments,
+var CollectDeploymentsMeta = plugin.SubTaskMeta{
+ Name: "CollectDeployments",
+ EntryPoint: CollectDeployments,
EnabledByDefault: true,
Description: "collect and extract github deployments to raw and tool layer from GithubGraphql api",
DomainTypes: []string{plugin.DOMAIN_TYPE_CICD},
@@ -85,9 +83,9 @@ type GraphqlQueryDeploymentDeployment struct {
UpdatedAt time.Time
}
-// CollectAndExtractDeployments will request github api via graphql and store the result into raw layer by default
+// CollectDeployments will request github api via graphql and store the result into raw layer by default
// ResponseParser's return will be stored to tool layer. So it's called CollectorAndExtractor.
-func CollectAndExtractDeployments(taskCtx plugin.SubTaskContext) errors.Error {
+func CollectDeployments(taskCtx plugin.SubTaskContext) errors.Error {
data := taskCtx.GetData().(*githubTasks.GithubTaskData)
collectorWithState, err := helper.NewStatefulApiCollector(helper.RawDataSubTaskArgs{
Ctx: taskCtx,
@@ -100,10 +98,6 @@ func CollectAndExtractDeployments(taskCtx plugin.SubTaskContext) errors.Error {
if err != nil {
return err
}
- since := helper.DateTime{}
- if collectorWithState.Since != nil {
- since = helper.DateTime{Time: *collectorWithState.Since}
- }
err = collectorWithState.InitGraphQLCollector(helper.GraphqlCollectorArgs{
GraphqlClient: data.GraphqlClient,
@@ -128,55 +122,11 @@ func CollectAndExtractDeployments(taskCtx plugin.SubTaskContext) errors.Error {
return query.Repository.Deployments.PageInfo, nil
},
ResponseParser: func(iQuery interface{}, variables map[string]interface{}) ([]interface{}, error) {
- query := iQuery.(*GraphqlQueryDeploymentWrapper)
- deployments := query.Repository.Deployments.Deployments
- var results []interface{}
- for _, deployment := range deployments {
- //Skip deployments with createdAt earlier than 'since'
- if deployment.CreatedAt.Before(since.Time) {
- continue
- }
- githubDeployment, err := convertGithubDeployment(deployment, data.Options.ConnectionId, data.Options.GithubId)
- if err != nil {
- return nil, err
- }
- results = append(results, githubDeployment)
- }
-
- return results, nil
+ return nil, nil
},
})
if err != nil {
return err
}
-
return collectorWithState.Execute()
}
-
-func convertGithubDeployment(deployment GraphqlQueryDeploymentDeployment, connectionId uint64, githubId int) (*githubModels.GithubDeployment, error) {
- ret := &githubModels.GithubDeployment{
- ConnectionId: connectionId,
- GithubId: githubId,
- NoPKModel: common.NewNoPKModel(),
- Id: deployment.Id,
- DatabaseId: deployment.DatabaseId,
- Payload: deployment.Payload,
- Description: deployment.Description,
- CommitOid: deployment.CommitOid,
- Environment: deployment.Environment,
- State: deployment.State,
- RepositoryID: deployment.Repository.Id,
- RepositoryName: deployment.Repository.Name,
- RepositoryUrl: deployment.Repository.Url,
- CreatedDate: deployment.CreatedAt,
- UpdatedDate: deployment.UpdatedAt,
- LatestStatusState: deployment.LatestStatus.State,
- }
- if !deployment.LatestStatus.UpdatedAt.IsZero() {
- ret.LatestUpdatedDate = deployment.LatestStatus.UpdatedAt
- }
- if deployment.Ref != nil {
- ret.RefName = deployment.Ref.Name
- }
- return ret, nil
-}
diff --git a/backend/plugins/github_graphql/tasks/deployment_extractor.go b/backend/plugins/github_graphql/tasks/deployment_extractor.go
new file mode 100644
index 000000000..730f07b75
--- /dev/null
+++ b/backend/plugins/github_graphql/tasks/deployment_extractor.go
@@ -0,0 +1,104 @@
+/*
+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"
+ githubModels "github.com/apache/incubator-devlake/plugins/github/models"
+ githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
+)
+
+var _ plugin.SubTaskEntryPoint = ExtractDeployments
+
+var ExtractDeploymentsMeta = plugin.SubTaskMeta{
+ Name: "extractDeployments",
+ EntryPoint: ExtractDeployments,
+ EnabledByDefault: true,
+ Description: "extract raw deployment data into tool layer table github_graphql_deployment",
+ DomainTypes: []string{plugin.DOMAIN_TYPE_CICD},
+}
+
+func ExtractDeployments(taskCtx plugin.SubTaskContext) errors.Error {
+ data := taskCtx.GetData().(*githubTasks.GithubTaskData)
+ extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+ RawDataSubTaskArgs: api.RawDataSubTaskArgs{
+ Ctx: taskCtx,
+ Params: githubTasks.GithubApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ Name: data.Options.Name,
+ },
+ Table: RAW_DEPLOYMENT,
+ },
+ Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+ apiDeployment := &GraphqlQueryDeploymentWrapper{}
+ err := errors.Convert(json.Unmarshal(row.Data, apiDeployment))
+ if err != nil {
+ return nil, err
+ }
+
+ deployments := apiDeployment.Repository.Deployments.Deployments
+ var results []interface{}
+ for _, deployment := range deployments {
+ githubDeployment, err := convertGithubDeployment(deployment, data.Options.ConnectionId, data.Options.GithubId)
+ if err != nil {
+ return nil, errors.Convert(err)
+ }
+ results = append(results, githubDeployment)
+ }
+
+ return results, nil
+ },
+ })
+ if err != nil {
+ return err
+ }
+ return extractor.Execute()
+}
+
+func convertGithubDeployment(deployment GraphqlQueryDeploymentDeployment, connectionId uint64, githubId int) (*githubModels.GithubDeployment, error) {
+ ret := &githubModels.GithubDeployment{
+ ConnectionId: connectionId,
+ GithubId: githubId,
+ NoPKModel: common.NewNoPKModel(),
+ Id: deployment.Id,
+ DatabaseId: deployment.DatabaseId,
+ Payload: deployment.Payload,
+ Description: deployment.Description,
+ CommitOid: deployment.CommitOid,
+ Environment: deployment.Environment,
+ State: deployment.State,
+ RepositoryID: deployment.Repository.Id,
+ RepositoryName: deployment.Repository.Name,
+ RepositoryUrl: deployment.Repository.Url,
+ CreatedDate: deployment.CreatedAt,
+ UpdatedDate: deployment.UpdatedAt,
+ LatestStatusState: deployment.LatestStatus.State,
+ }
+ if !deployment.LatestStatus.UpdatedAt.IsZero() {
+ ret.LatestUpdatedDate = deployment.LatestStatus.UpdatedAt
+ }
+ if deployment.Ref != nil {
+ ret.RefName = deployment.Ref.Name
+ }
+ return ret, nil
+}
diff --git a/backend/plugins/github_graphql/tasks/issue_collector.go b/backend/plugins/github_graphql/tasks/issue_collector.go
index cfb0459e4..035576002 100644
--- a/backend/plugins/github_graphql/tasks/issue_collector.go
+++ b/backend/plugins/github_graphql/tasks/issue_collector.go
@@ -21,12 +21,9 @@ import (
"strings"
"time"
- "github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
- "github.com/apache/incubator-devlake/core/models/domainlayer/ticket"
"github.com/apache/incubator-devlake/core/plugin"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
- "github.com/apache/incubator-devlake/plugins/github/models"
githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
"github.com/merico-dev/graphql"
)
@@ -73,30 +70,18 @@ type GraphqlQueryIssue struct {
} `graphql:"labels(first: 100)"`
}
-var CollectIssueMeta = plugin.SubTaskMeta{
- Name: "CollectIssue",
- EntryPoint: CollectIssue,
+var CollectIssuesMeta = plugin.SubTaskMeta{
+ Name: "CollectIssues",
+ EntryPoint: CollectIssues,
EnabledByDefault: true,
Description: "Collect Issue data from GithubGraphql api, supports both timeFilter and diffSync.",
DomainTypes: []string{plugin.DOMAIN_TYPE_TICKET},
}
-var _ plugin.SubTaskEntryPoint = CollectIssue
+var _ plugin.SubTaskEntryPoint = CollectIssues
-func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
- db := taskCtx.GetDal()
+func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error {
data := taskCtx.GetData().(*githubTasks.GithubTaskData)
- config := data.Options.ScopeConfig
- issueRegexes, err := githubTasks.NewIssueRegexes(config)
- if err != nil {
- return nil
- }
-
- milestoneMap, err := getMilestoneMap(db, data.Options.GithubId, data.Options.ConnectionId)
- if err != nil {
- return nil
- }
-
collectorWithState, err := helper.NewStatefulApiCollector(helper.RawDataSubTaskArgs{
Ctx: taskCtx,
Params: githubTasks.GithubApiParams{
@@ -136,42 +121,7 @@ func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
return query.Repository.IssueList.PageInfo, nil
},
ResponseParser: func(iQuery interface{}, variables map[string]interface{}) ([]interface{}, error) {
- query := iQuery.(*GraphqlQueryIssueWrapper)
- issues := query.Repository.IssueList.Issues
-
- results := make([]interface{}, 0, 1)
- isFinish := false
- for _, issue := range issues {
- githubIssue, err := convertGithubIssue(milestoneMap, issue, data.Options.ConnectionId, data.Options.GithubId)
- if err != nil {
- return nil, err
- }
- githubLabels, err := convertGithubLabels(issueRegexes, issue, githubIssue)
- if err != nil {
- return nil, err
- }
- results = append(results, githubLabels...)
- results = append(results, githubIssue)
- if len(issue.AssigneeList.Assignees) > 0 {
- extractGraphqlPreAccount(&results, &issue.AssigneeList.Assignees[0], data.Options.GithubId, data.Options.ConnectionId)
- }
- extractGraphqlPreAccount(&results, issue.Author, data.Options.GithubId, data.Options.ConnectionId)
- for _, assignee := range issue.AssigneeList.Assignees {
- issueAssignee := &models.GithubIssueAssignee{
- ConnectionId: githubIssue.ConnectionId,
- IssueId: githubIssue.GithubId,
- RepoId: githubIssue.RepoId,
- AssigneeId: assignee.Id,
- AssigneeName: assignee.Login,
- }
- results = append(results, issueAssignee)
- }
- }
- if isFinish {
- return results, helper.ErrFinishCollect
- } else {
- return results, nil
- }
+ return nil, nil
},
})
if err != nil {
@@ -180,101 +130,3 @@ func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
return collectorWithState.Execute()
}
-
-// create a milestone map for numberId to databaseId
-func getMilestoneMap(db dal.Dal, repoId int, connectionId uint64) (map[int]int, errors.Error) {
- milestoneMap := map[int]int{}
- var milestones []struct {
- MilestoneId int
- RepoId int
- Number int
- }
- err := db.All(
- &milestones,
- dal.From(&models.GithubMilestone{}),
- dal.Where("repo_id = ? and connection_id = ?", repoId, connectionId),
- )
- if err != nil {
- return nil, err
- }
- for _, milestone := range milestones {
- milestoneMap[milestone.Number] = milestone.MilestoneId
- }
- return milestoneMap, nil
-}
-
-func convertGithubIssue(milestoneMap map[int]int, issue GraphqlQueryIssue, connectionId uint64, repositoryId int) (*models.GithubIssue, errors.Error) {
- githubIssue := &models.GithubIssue{
- ConnectionId: connectionId,
- GithubId: issue.DatabaseId,
- RepoId: repositoryId,
- Number: issue.Number,
- State: issue.State,
- Title: issue.Title,
- Body: strings.ReplaceAll(issue.Body, "\x00", `<0x00>`),
- Url: issue.Url,
- ClosedAt: issue.ClosedAt,
- GithubCreatedAt: issue.CreatedAt,
- GithubUpdatedAt: issue.UpdatedAt,
- }
- if issue.AssigneeList.Assignees != nil && len(issue.AssigneeList.Assignees) > 0 {
- githubIssue.AssigneeId = issue.AssigneeList.Assignees[0].Id
- githubIssue.AssigneeName = issue.AssigneeList.Assignees[0].Login
- }
- if issue.Author != nil {
- githubIssue.AuthorId = issue.Author.Id
- githubIssue.AuthorName = issue.Author.Login
- }
- if issue.ClosedAt != nil {
- githubIssue.LeadTimeMinutes = uint(issue.ClosedAt.Sub(issue.CreatedAt).Minutes())
- }
- if issue.Milestone != nil {
- if milestoneId, ok := milestoneMap[issue.Milestone.Number]; ok {
- githubIssue.MilestoneId = milestoneId
- }
- }
- return githubIssue, nil
-}
-
-func convertGithubLabels(issueRegexes *githubTasks.IssueRegexes, issue GraphqlQueryIssue, githubIssue *models.GithubIssue) ([]interface{}, errors.Error) {
- var results []interface{}
- var joinedLabels []string
- for _, label := range issue.Labels.Nodes {
- results = append(results, &models.GithubIssueLabel{
- ConnectionId: githubIssue.ConnectionId,
- IssueId: githubIssue.GithubId,
- LabelName: label.Name,
- })
- joinedLabels = append(joinedLabels, label.Name)
-
- if issueRegexes.SeverityRegex != nil {
- groups := issueRegexes.SeverityRegex.FindStringSubmatch(label.Name)
- if len(groups) > 1 {
- githubIssue.Severity = groups[1]
- }
- }
- if issueRegexes.ComponentRegex != nil {
- groups := issueRegexes.ComponentRegex.FindStringSubmatch(label.Name)
- if len(groups) > 1 {
- githubIssue.Component = groups[1]
- }
- }
- if issueRegexes.PriorityRegex != nil {
- groups := issueRegexes.PriorityRegex.FindStringSubmatch(label.Name)
- if len(groups) > 1 {
- githubIssue.Priority = groups[1]
- }
- }
- if issueRegexes.TypeRequirementRegex != nil && issueRegexes.TypeRequirementRegex.MatchString(label.Name) {
- githubIssue.StdType = ticket.REQUIREMENT
- } else if issueRegexes.TypeBugRegex != nil && issueRegexes.TypeBugRegex.MatchString(label.Name) {
- githubIssue.StdType = ticket.BUG
- } else if issueRegexes.TypeIncidentRegex != nil && issueRegexes.TypeIncidentRegex.MatchString(label.Name) {
- githubIssue.StdType = ticket.INCIDENT
- }
- }
- if len(joinedLabels) > 0 {
- githubIssue.Type = strings.Join(joinedLabels, ",")
- }
- return results, nil
-}
diff --git a/backend/plugins/github_graphql/tasks/issue_collector.go b/backend/plugins/github_graphql/tasks/issue_extractor.go
similarity index 65%
copy from backend/plugins/github_graphql/tasks/issue_collector.go
copy to backend/plugins/github_graphql/tasks/issue_extractor.go
index cfb0459e4..3dc61b418 100644
--- a/backend/plugins/github_graphql/tasks/issue_collector.go
+++ b/backend/plugins/github_graphql/tasks/issue_extractor.go
@@ -18,72 +18,29 @@ limitations under the License.
package tasks
import (
+ "encoding/json"
"strings"
- "time"
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/models/domainlayer/ticket"
"github.com/apache/incubator-devlake/core/plugin"
- helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+ "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/github/models"
githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
- "github.com/merico-dev/graphql"
)
-const RAW_ISSUES_TABLE = "github_graphql_issues"
+var _ plugin.SubTaskEntryPoint = ExtractAccounts
-type GraphqlQueryIssueWrapper struct {
- RateLimit struct {
- Cost int
- }
- Repository struct {
- IssueList struct {
- TotalCount graphql.Int
- Issues []GraphqlQueryIssue `graphql:"nodes"`
- PageInfo *helper.GraphqlQueryPageInfo
- } `graphql:"issues(first: $pageSize, after: $skipCursor, orderBy: {field: CREATED_AT, direction: DESC}, filterBy: {since: $since})"`
- } `graphql:"repository(owner: $owner, name: $name)"`
-}
-
-type GraphqlQueryIssue struct {
- DatabaseId int
- Number int
- State string
- StateReason string
- Title string
- Body string
- Author *GraphqlInlineAccountQuery
- Url string
- ClosedAt *time.Time
- CreatedAt time.Time
- UpdatedAt time.Time
- AssigneeList struct {
- // FIXME now domain layer just support one assignee
- Assignees []GraphqlInlineAccountQuery `graphql:"nodes"`
- } `graphql:"assignees(first: 100)"`
- Milestone *struct {
- Number int
- } `json:"milestone"`
- Labels struct {
- Nodes []struct {
- Id string
- Name string
- }
- } `graphql:"labels(first: 100)"`
-}
-
-var CollectIssueMeta = plugin.SubTaskMeta{
- Name: "CollectIssue",
- EntryPoint: CollectIssue,
+var ExtractIssuesMeta = plugin.SubTaskMeta{
+ Name: "extractIssues",
+ EntryPoint: ExtractIssues,
EnabledByDefault: true,
- Description: "Collect Issue data from GithubGraphql api, supports both timeFilter and diffSync.",
+ Description: "Extract raw Issues data into tool layer table github_issues",
DomainTypes: []string{plugin.DOMAIN_TYPE_TICKET},
}
-var _ plugin.SubTaskEntryPoint = CollectIssue
-
-func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
+func ExtractIssues(taskCtx plugin.SubTaskContext) errors.Error {
db := taskCtx.GetDal()
data := taskCtx.GetData().(*githubTasks.GithubTaskData)
config := data.Options.ScopeConfig
@@ -91,56 +48,28 @@ func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
if err != nil {
return nil
}
-
milestoneMap, err := getMilestoneMap(db, data.Options.GithubId, data.Options.ConnectionId)
if err != nil {
return nil
}
- collectorWithState, err := helper.NewStatefulApiCollector(helper.RawDataSubTaskArgs{
- Ctx: taskCtx,
- Params: githubTasks.GithubApiParams{
- ConnectionId: data.Options.ConnectionId,
- Name: data.Options.Name,
+ extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+ RawDataSubTaskArgs: api.RawDataSubTaskArgs{
+ Ctx: taskCtx,
+ Params: githubTasks.GithubApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ Name: data.Options.Name,
+ },
+ Table: RAW_ISSUES_TABLE,
},
- Table: RAW_ISSUES_TABLE,
- })
- if err != nil {
- return err
- }
-
- err = collectorWithState.InitGraphQLCollector(helper.GraphqlCollectorArgs{
- GraphqlClient: data.GraphqlClient,
- PageSize: 100,
- BuildQuery: func(reqData *helper.GraphqlRequestData) (interface{}, map[string]interface{}, error) {
- query := &GraphqlQueryIssueWrapper{}
- if reqData == nil {
- return query, map[string]interface{}{}, nil
- }
- since := helper.DateTime{}
- if collectorWithState.Since != nil {
- since = helper.DateTime{Time: *collectorWithState.Since}
- }
- ownerName := strings.Split(data.Options.Name, "/")
- variables := map[string]interface{}{
- "since": since,
- "pageSize": graphql.Int(reqData.Pager.Size),
- "skipCursor": (*graphql.String)(reqData.Pager.SkipCursor),
- "owner": graphql.String(ownerName[0]),
- "name": graphql.String(ownerName[1]),
+ Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+ apiIssue := &GraphqlQueryIssueWrapper{}
+ err := errors.Convert(json.Unmarshal(row.Data, apiIssue))
+ if err != nil {
+ return nil, err
}
- return query, variables, nil
- },
- GetPageInfo: func(iQuery interface{}, args *helper.GraphqlCollectorArgs) (*helper.GraphqlQueryPageInfo, error) {
- query := iQuery.(*GraphqlQueryIssueWrapper)
- return query.Repository.IssueList.PageInfo, nil
- },
- ResponseParser: func(iQuery interface{}, variables map[string]interface{}) ([]interface{}, error) {
- query := iQuery.(*GraphqlQueryIssueWrapper)
- issues := query.Repository.IssueList.Issues
-
+ issues := apiIssue.Repository.IssueList.Issues
results := make([]interface{}, 0, 1)
- isFinish := false
for _, issue := range issues {
githubIssue, err := convertGithubIssue(milestoneMap, issue, data.Options.ConnectionId, data.Options.GithubId)
if err != nil {
@@ -167,18 +96,15 @@ func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
results = append(results, issueAssignee)
}
}
- if isFinish {
- return results, helper.ErrFinishCollect
- } else {
- return results, nil
- }
+ return results, nil
},
})
+
if err != nil {
return err
}
- return collectorWithState.Execute()
+ return extractor.Execute()
}
// create a milestone map for numberId to databaseId
diff --git a/backend/plugins/github_graphql/tasks/job_collector.go b/backend/plugins/github_graphql/tasks/job_collector.go
index 7c838f980..650087de3 100644
--- a/backend/plugins/github_graphql/tasks/job_collector.go
+++ b/backend/plugins/github_graphql/tasks/job_collector.go
@@ -18,14 +18,11 @@ limitations under the License.
package tasks
import (
- "encoding/json"
"reflect"
- "strings"
"time"
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
- "github.com/apache/incubator-devlake/core/models/domainlayer/devops"
"github.com/apache/incubator-devlake/core/plugin"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/github/models"
@@ -89,9 +86,9 @@ type SimpleWorkflowRun struct {
CheckSuiteNodeID string
}
-var CollectGraphqlJobsMeta = plugin.SubTaskMeta{
- Name: "CollectGraphqlJobs",
- EntryPoint: CollectGraphqlJobs,
+var CollectJobsMeta = plugin.SubTaskMeta{
+ Name: "CollectJobs",
+ EntryPoint: CollectJobs,
EnabledByDefault: true,
Description: "Collect Jobs(CheckRun) data from GithubGraphql api, supports both timeFilter and diffSync.",
DomainTypes: []string{plugin.DOMAIN_TYPE_CICD},
@@ -99,8 +96,7 @@ var CollectGraphqlJobsMeta = plugin.SubTaskMeta{
var _ plugin.SubTaskEntryPoint = CollectAccount
-func CollectGraphqlJobs(taskCtx plugin.SubTaskContext) errors.Error {
- logger := taskCtx.GetLogger()
+func CollectJobs(taskCtx plugin.SubTaskContext) errors.Error {
db := taskCtx.GetDal()
data := taskCtx.GetData().(*githubTasks.GithubTaskData)
@@ -165,45 +161,7 @@ func CollectGraphqlJobs(taskCtx plugin.SubTaskContext) errors.Error {
// log and ignore
taskCtx.GetLogger().Warn(dataError, `query check run get error but ignore`)
}
- query := iQuery.(*GraphqlQueryCheckRunWrapper)
- nodes := query.Node
-
- results := make([]interface{}, 0, 1)
- for _, node := range nodes {
- for _, checkRun := range node.CheckSuite.CheckRuns.Nodes {
-
- paramsBytes, err := json.Marshal(checkRun.Steps.Nodes)
- if err != nil {
- logger.Error(err, `Marshal checkRun.Steps.Nodes fail and ignore`)
- }
- githubJob := &models.GithubJob{
- ConnectionId: data.Options.ConnectionId,
- RunID: node.CheckSuite.WorkflowRun.DatabaseId,
- RepoId: data.Options.GithubId,
- ID: checkRun.DatabaseId,
- NodeID: checkRun.Id,
- HTMLURL: checkRun.DetailsUrl,
- Status: strings.ToUpper(checkRun.Status),
- Conclusion: strings.ToUpper(checkRun.Conclusion),
- StartedAt: checkRun.StartedAt,
- CompletedAt: checkRun.CompletedAt,
- Name: checkRun.Name,
- Steps: paramsBytes,
- Type: data.RegexEnricher.ReturnNameIfMatched(devops.DEPLOYMENT, checkRun.Name),
- Environment: data.RegexEnricher.ReturnNameIfOmittedOrMatched(devops.PRODUCTION, checkRun.Name),
- // these columns can not fill by graphql
- //HeadSha: ``, // use _tool_github_runs
- //RunURL: ``,
- //CheckRunURL: ``,
- //Labels: ``, // not on use
- //RunnerID: ``, // not on use
- //RunnerName: ``, // not on use
- //RunnerGroupID: ``, // not on use
- }
- results = append(results, githubJob)
- }
- }
- return results, nil
+ return nil, nil
},
})
if err != nil {
diff --git a/backend/plugins/github_graphql/tasks/job_extractor.go b/backend/plugins/github_graphql/tasks/job_extractor.go
new file mode 100644
index 000000000..73a4c486c
--- /dev/null
+++ b/backend/plugins/github_graphql/tasks/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"
+ "strings"
+
+ "github.com/apache/incubator-devlake/core/errors"
+ "github.com/apache/incubator-devlake/core/models/domainlayer/devops"
+ "github.com/apache/incubator-devlake/core/plugin"
+ "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+ "github.com/apache/incubator-devlake/plugins/github/models"
+ githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
+)
+
+var _ plugin.SubTaskEntryPoint = ExtractAccounts
+
+var ExtractJobsMeta = plugin.SubTaskMeta{
+ Name: "extractJobs",
+ EntryPoint: ExtractJobs,
+ EnabledByDefault: true,
+ Description: "Extract raw run data into tool layer table github_jobs",
+ DomainTypes: []string{plugin.DOMAIN_TYPE_CICD},
+}
+
+func ExtractJobs(taskCtx plugin.SubTaskContext) errors.Error {
+ data := taskCtx.GetData().(*githubTasks.GithubTaskData)
+ extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+ RawDataSubTaskArgs: api.RawDataSubTaskArgs{
+ Ctx: taskCtx,
+ Params: githubTasks.GithubApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ Name: data.Options.Name,
+ },
+ Table: RAW_GRAPHQL_JOBS_TABLE,
+ },
+ Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+ apiJob := &GraphqlQueryCheckRunWrapper{}
+ err := errors.Convert(json.Unmarshal(row.Data, apiJob))
+ if err != nil {
+ return nil, err
+ }
+
+ nodes := apiJob.Node
+ results := make([]interface{}, 0, 1)
+ for _, node := range nodes {
+ for _, checkRun := range node.CheckSuite.CheckRuns.Nodes {
+
+ paramsBytes, err := json.Marshal(checkRun.Steps.Nodes)
+ if err != nil {
+ taskCtx.GetLogger().Error(err, `Marshal checkRun.Steps.Nodes fail and ignore`)
+ }
+ githubJob := &models.GithubJob{
+ ConnectionId: data.Options.ConnectionId,
+ RunID: node.CheckSuite.WorkflowRun.DatabaseId,
+ RepoId: data.Options.GithubId,
+ ID: checkRun.DatabaseId,
+ NodeID: checkRun.Id,
+ HTMLURL: checkRun.DetailsUrl,
+ Status: strings.ToUpper(checkRun.Status),
+ Conclusion: strings.ToUpper(checkRun.Conclusion),
+ StartedAt: checkRun.StartedAt,
+ CompletedAt: checkRun.CompletedAt,
+ Name: checkRun.Name,
+ Steps: paramsBytes,
+ Type: data.RegexEnricher.ReturnNameIfMatched(devops.DEPLOYMENT, checkRun.Name),
+ Environment: data.RegexEnricher.ReturnNameIfOmittedOrMatched(devops.PRODUCTION, checkRun.Name),
+ // these columns can not fill by graphql
+ //HeadSha: ``, // use _tool_github_runs
+ //RunURL: ``,
+ //CheckRunURL: ``,
+ //Labels: ``, // not on use
+ //RunnerID: ``, // not on use
+ //RunnerName: ``, // not on use
+ //RunnerGroupID: ``, // not on use
+ }
+ results = append(results, githubJob)
+ }
+ }
+ return results, nil
+
+ },
+ })
+
+ if err != nil {
+ return err
+ }
+
+ return extractor.Execute()
+}
diff --git a/backend/plugins/github_graphql/tasks/pr_collector.go b/backend/plugins/github_graphql/tasks/pr_collector.go
index 37d10e742..152ea27b2 100644
--- a/backend/plugins/github_graphql/tasks/pr_collector.go
+++ b/backend/plugins/github_graphql/tasks/pr_collector.go
@@ -18,14 +18,12 @@ limitations under the License.
package tasks
import (
- "regexp"
"strings"
"time"
"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/github/models"
"github.com/apache/incubator-devlake/plugins/github/tasks"
"github.com/merico-dev/graphql"
)
@@ -119,35 +117,19 @@ type GraphqlQueryCommit struct {
Url string
}
-var CollectPrMeta = plugin.SubTaskMeta{
- Name: "CollectPr",
- EntryPoint: CollectPr,
+var CollectPrsMeta = plugin.SubTaskMeta{
+ Name: "CollectPrs",
+ EntryPoint: CollectPrs,
EnabledByDefault: true,
Description: "Collect Pr data from GithubGraphql api, supports both timeFilter and diffSync.",
DomainTypes: []string{plugin.DOMAIN_TYPE_CODE_REVIEW},
}
-var _ plugin.SubTaskEntryPoint = CollectPr
+var _ plugin.SubTaskEntryPoint = CollectPrs
-func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
+func CollectPrs(taskCtx plugin.SubTaskContext) errors.Error {
data := taskCtx.GetData().(*tasks.GithubTaskData)
- config := data.Options.ScopeConfig
- var labelTypeRegex *regexp.Regexp
- var labelComponentRegex *regexp.Regexp
var err errors.Error
- if config != nil && len(config.PrType) > 0 {
- labelTypeRegex, err = errors.Convert01(regexp.Compile(config.PrType))
- if err != nil {
- return errors.Default.Wrap(err, "regexp Compile prType failed")
- }
- }
- if config != nil && len(config.PrComponent) > 0 {
- labelComponentRegex, err = errors.Convert01(regexp.Compile(config.PrComponent))
- if err != nil {
- return errors.Default.Wrap(err, "regexp Compile prComponent failed")
- }
- }
-
collectorWithState, err := api.NewStatefulApiCollector(api.RawDataSubTaskArgs{
Ctx: taskCtx,
Params: tasks.GithubApiParams{
@@ -188,7 +170,6 @@ func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
query := iQuery.(*GraphqlQueryPrWrapper)
prs := query.Repository.PullRequests.Prs
- results := make([]interface{}, 0, 1)
isFinish := false
for _, rawL := range prs {
// collect data even though in increment mode because of updating existing data
@@ -196,85 +177,11 @@ func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
isFinish = true
break
}
- githubPr, err := convertGithubPullRequest(rawL, data.Options.ConnectionId, data.Options.GithubId)
- if err != nil {
- return nil, err
- }
- extractGraphqlPreAccount(&results, rawL.Author, data.Options.GithubId, data.Options.ConnectionId)
- for _, label := range rawL.Labels.Nodes {
- results = append(results, &models.GithubPrLabel{
- ConnectionId: data.Options.ConnectionId,
- PullId: githubPr.GithubId,
- LabelName: label.Name,
- })
- // if pr.Type has not been set and prType is set in .env, process the below
- if labelTypeRegex != nil {
- groups := labelTypeRegex.FindStringSubmatch(label.Name)
- if len(groups) > 0 {
- githubPr.Type = groups[1]
- }
- }
-
- // if pr.Component has not been set and prComponent is set in .env, process
- if labelComponentRegex != nil {
- groups := labelComponentRegex.FindStringSubmatch(label.Name)
- if len(groups) > 0 {
- githubPr.Component = groups[1]
- }
- }
- }
- results = append(results, githubPr)
-
- for _, apiPullRequestReview := range rawL.Reviews.Nodes {
- if apiPullRequestReview.State != "PENDING" {
- githubPrReview := &models.GithubPrReview{
- ConnectionId: data.Options.ConnectionId,
- GithubId: apiPullRequestReview.DatabaseId,
- Body: apiPullRequestReview.Body,
- State: apiPullRequestReview.State,
- CommitSha: apiPullRequestReview.Commit.Oid,
- GithubSubmitAt: apiPullRequestReview.SubmittedAt,
-
- PullRequestId: githubPr.GithubId,
- }
-
- if apiPullRequestReview.Author != nil {
- githubPrReview.AuthorUserId = apiPullRequestReview.Author.Id
- githubPrReview.AuthorUsername = apiPullRequestReview.Author.Login
- extractGraphqlPreAccount(&results, apiPullRequestReview.Author, data.Options.GithubId, data.Options.ConnectionId)
- }
-
- results = append(results, githubPrReview)
- }
- }
-
- for _, apiPullRequestCommit := range rawL.Commits.Nodes {
- githubCommit, err := convertPullRequestCommit(apiPullRequestCommit)
- if err != nil {
- return nil, err
- }
- results = append(results, githubCommit)
-
- githubPullRequestCommit := &models.GithubPrCommit{
- ConnectionId: data.Options.ConnectionId,
- CommitSha: apiPullRequestCommit.Commit.Oid,
- PullRequestId: githubPr.GithubId,
- CommitAuthorName: githubCommit.AuthorName,
- CommitAuthorEmail: githubCommit.AuthorEmail,
- CommitAuthoredDate: githubCommit.AuthoredDate,
- }
- if err != nil {
- return nil, err
- }
- results = append(results, githubPullRequestCommit)
- extractGraphqlPreAccount(&results, apiPullRequestCommit.Commit.Author.User, data.Options.GithubId, data.Options.ConnectionId)
- }
}
-
if isFinish {
- return results, api.ErrFinishCollect
+ return nil, api.ErrFinishCollect
} else {
- return results, nil
+ return nil, nil
}
},
})
@@ -284,50 +191,3 @@ func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
return collectorWithState.Execute()
}
-
-func convertGithubPullRequest(pull GraphqlQueryPr, connId uint64, repoId int) (*models.GithubPullRequest, errors.Error) {
- githubPull := &models.GithubPullRequest{
- ConnectionId: connId,
- GithubId: pull.DatabaseId,
- RepoId: repoId,
- Number: pull.Number,
- State: pull.State,
- Title: pull.Title,
- Url: pull.Url,
- GithubCreatedAt: pull.CreatedAt,
- GithubUpdatedAt: pull.UpdatedAt,
- ClosedAt: pull.ClosedAt,
- MergedAt: pull.MergedAt,
- Body: pull.Body,
- BaseRef: pull.BaseRefName,
- BaseCommitSha: pull.BaseRefOid,
- HeadRef: pull.HeadRefName,
- HeadCommitSha: pull.HeadRefOid,
- }
- if pull.MergeCommit != nil {
- githubPull.MergeCommitSha = pull.MergeCommit.Oid
- }
- if pull.Author != nil {
- githubPull.AuthorName = pull.Author.Login
- githubPull.AuthorId = pull.Author.Id
- }
- return githubPull, nil
-}
-
-func convertPullRequestCommit(prCommit GraphqlQueryCommit) (*models.GithubCommit, errors.Error) {
- githubCommit := &models.GithubCommit{
- Sha: prCommit.Commit.Oid,
- Message: prCommit.Commit.Message,
- AuthorName: prCommit.Commit.Author.Name,
- AuthorEmail: prCommit.Commit.Author.Email,
- AuthoredDate: prCommit.Commit.Author.Date,
- CommitterName: prCommit.Commit.Committer.Name,
- CommitterEmail: prCommit.Commit.Committer.Email,
- CommittedDate: prCommit.Commit.Committer.Date,
- Url: prCommit.Url,
- }
- if prCommit.Commit.Author.User != nil {
- githubCommit.AuthorId = prCommit.Commit.Author.User.Id
- }
- return githubCommit, nil
-}
diff --git a/backend/plugins/github_graphql/tasks/pr_collector.go b/backend/plugins/github_graphql/tasks/pr_extractor.go
similarity index 57%
copy from backend/plugins/github_graphql/tasks/pr_collector.go
copy to backend/plugins/github_graphql/tasks/pr_extractor.go
index 37d10e742..9941e48a0 100644
--- a/backend/plugins/github_graphql/tasks/pr_collector.go
+++ b/backend/plugins/github_graphql/tasks/pr_extractor.go
@@ -18,123 +18,33 @@ limitations under the License.
package tasks
import (
+ "encoding/json"
"regexp"
- "strings"
- "time"
"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/github/models"
- "github.com/apache/incubator-devlake/plugins/github/tasks"
- "github.com/merico-dev/graphql"
+ githubTasks "github.com/apache/incubator-devlake/plugins/github/tasks"
)
-const RAW_PRS_TABLE = "github_graphql_prs"
+var _ plugin.SubTaskEntryPoint = ExtractPrs
-type GraphqlQueryPrWrapper struct {
- RateLimit struct {
- Cost int
- }
- // now it orderBy UPDATED_AT and use cursor pagination
- // It may miss some PRs updated when collection.
- // Because these missed PRs will be collected on next, But it's not enough.
- // So Next Millstone(0.17) we should change it to filter by CREATE_AT + collect detail
- Repository struct {
- PullRequests struct {
- PageInfo *api.GraphqlQueryPageInfo
- Prs []GraphqlQueryPr `graphql:"nodes"`
- TotalCount graphql.Int
- } `graphql:"pullRequests(first: $pageSize, after: $skipCursor, orderBy: {field: UPDATED_AT, direction: DESC})"`
- } `graphql:"repository(owner: $owner, name: $name)"`
-}
-
-type GraphqlQueryPr struct {
- DatabaseId int
- Number int
- State string
- Title string
- Body string
- Url string
- Labels struct {
- Nodes []struct {
- Id string
- Name string
- }
- } `graphql:"labels(first: 100)"`
- Author *GraphqlInlineAccountQuery
- Assignees struct {
- // FIXME now domain layer just support one assignee
- Assignees []GraphqlInlineAccountQuery `graphql:"nodes"`
- } `graphql:"assignees(first: 1)"`
- ClosedAt *time.Time
- MergedAt *time.Time
- UpdatedAt time.Time
- CreatedAt time.Time
- MergeCommit *struct {
- Oid string
- }
- HeadRefName string
- HeadRefOid string
- BaseRefName string
- BaseRefOid string
- Commits struct {
- PageInfo *api.GraphqlQueryPageInfo
- Nodes []GraphqlQueryCommit `graphql:"nodes"`
- TotalCount graphql.Int
- } `graphql:"commits(first: 100)"`
- Reviews struct {
- TotalCount graphql.Int
- Nodes []GraphqlQueryReview `graphql:"nodes"`
- } `graphql:"reviews(first: 100)"`
-}
-
-type GraphqlQueryReview struct {
- Body string
- Author *GraphqlInlineAccountQuery
- State string `json:"state"`
- DatabaseId int `json:"databaseId"`
- Commit struct {
- Oid string
- }
- SubmittedAt *time.Time `json:"submittedAt"`
-}
-
-type GraphqlQueryCommit struct {
- Commit struct {
- Oid string
- Message string
- Author struct {
- Name string
- Email string
- Date time.Time
- User *GraphqlInlineAccountQuery
- }
- Committer struct {
- Date time.Time
- Email string
- Name string
- }
- }
- Url string
-}
-
-var CollectPrMeta = plugin.SubTaskMeta{
- Name: "CollectPr",
- EntryPoint: CollectPr,
+var ExtractPrsMeta = plugin.SubTaskMeta{
+ Name: "extractPrs",
+ EntryPoint: ExtractPrs,
EnabledByDefault: true,
- Description: "Collect Pr data from GithubGraphql api, supports both timeFilter and diffSync.",
- DomainTypes: []string{plugin.DOMAIN_TYPE_CODE_REVIEW},
+ Description: "Extract raw PullRequests data into tool layer table github_pull_requests",
+ DomainTypes: []string{plugin.DOMAIN_TYPE_CROSS},
}
-var _ plugin.SubTaskEntryPoint = CollectPr
-
-func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
- data := taskCtx.GetData().(*tasks.GithubTaskData)
+func ExtractPrs(taskCtx plugin.SubTaskContext) errors.Error {
+ data := taskCtx.GetData().(*githubTasks.GithubTaskData)
config := data.Options.ScopeConfig
var labelTypeRegex *regexp.Regexp
var labelComponentRegex *regexp.Regexp
var err errors.Error
+
if config != nil && len(config.PrType) > 0 {
labelTypeRegex, err = errors.Convert01(regexp.Compile(config.PrType))
if err != nil {
@@ -147,55 +57,25 @@ func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
return errors.Default.Wrap(err, "regexp Compile prComponent failed")
}
}
-
- collectorWithState, err := api.NewStatefulApiCollector(api.RawDataSubTaskArgs{
- Ctx: taskCtx,
- Params: tasks.GithubApiParams{
- ConnectionId: data.Options.ConnectionId,
- Name: data.Options.Name,
+ extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
+ RawDataSubTaskArgs: api.RawDataSubTaskArgs{
+ Ctx: taskCtx,
+ Params: githubTasks.GithubApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ Name: data.Options.Name,
+ },
+ Table: RAW_PRS_TABLE,
},
- Table: RAW_PRS_TABLE,
- })
- if err != nil {
- return err
- }
-
- err = collectorWithState.InitGraphQLCollector(api.GraphqlCollectorArgs{
- GraphqlClient: data.GraphqlClient,
- PageSize: 10,
- /*
- (Optional) Return query string for request, or you can plug them into UrlTemplate directly
- */
- BuildQuery: func(reqData *api.GraphqlRequestData) (interface{}, map[string]interface{}, error) {
- query := &GraphqlQueryPrWrapper{}
- if reqData == nil {
- return query, map[string]interface{}{}, nil
+ Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+ apiPr := &GraphqlQueryPrWrapper{}
+ err := errors.Convert(json.Unmarshal(row.Data, apiPr))
+ if err != nil {
+ return nil, err
}
- ownerName := strings.Split(data.Options.Name, "/")
- variables := map[string]interface{}{
- "pageSize": graphql.Int(reqData.Pager.Size),
- "skipCursor": (*graphql.String)(reqData.Pager.SkipCursor),
- "owner": graphql.String(ownerName[0]),
- "name": graphql.String(ownerName[1]),
- }
- return query, variables, nil
- },
- GetPageInfo: func(iQuery interface{}, args *api.GraphqlCollectorArgs) (*api.GraphqlQueryPageInfo, error) {
- query := iQuery.(*GraphqlQueryPrWrapper)
- return query.Repository.PullRequests.PageInfo, nil
- },
- ResponseParser: func(iQuery interface{}, variables map[string]interface{}) ([]interface{}, error) {
- query := iQuery.(*GraphqlQueryPrWrapper)
- prs := query.Repository.PullRequests.Prs
+ prs := apiPr.Repository.PullRequests.Prs
results := make([]interface{}, 0, 1)
- isFinish := false
for _, rawL := range prs {
- // collect data even though in increment mode because of updating existing data
- if collectorWithState.Since != nil && !collectorWithState.Since.Before(rawL.UpdatedAt) {
- isFinish = true
- break
- }
githubPr, err := convertGithubPullRequest(rawL, data.Options.ConnectionId, data.Options.GithubId)
if err != nil {
return nil, err
@@ -270,19 +150,15 @@ func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
extractGraphqlPreAccount(&results, apiPullRequestCommit.Commit.Author.User, data.Options.GithubId, data.Options.ConnectionId)
}
}
-
- if isFinish {
- return results, api.ErrFinishCollect
- } else {
- return results, nil
- }
+ return results, nil
},
})
+
if err != nil {
return err
}
- return collectorWithState.Execute()
+ return extractor.Execute()
}
func convertGithubPullRequest(pull GraphqlQueryPr, connId uint64, repoId int) (*models.GithubPullRequest, errors.Error) {