backend/helpers/pluginhelper/api/api_collector.go

/*
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 (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sync"
	"text/template"
	"time"

	"github.com/apache/incubator-devlake/core/plugin"

	"github.com/apache/incubator-devlake/core/dal"
	"github.com/apache/incubator-devlake/core/errors"
)

var _ plugin.SubTask = (*ApiCollector)(nil)

// Pager contains pagination information for a api request
type Pager struct {
	Page int
	Skip int
	Size int
}

// RequestData is the input of `UrlTemplate` `Query` and `Header`, so we can generate them dynamically
type RequestData struct {
	Pager     *Pager
	Params    interface{}
	Input     interface{}
	InputJSON []byte
	// equal to the return value from GetNextPageCustomData when PageSize>0 and not the first request
	CustomData interface{}
}

// AsyncResponseHandler FIXME ...
type AsyncResponseHandler func(res *http.Response) error

var ErrUndetermined = errors.BadInput.New("undetermined")

// ApiCollectorArgs FIXME ...
type ApiCollectorArgs struct {
	RawDataSubTaskArgs
	// UrlTemplate is used to generate the final URL for Api Collector to request
	// i.e. `api/3/issue/{{ .Input.IssueId }}/changelog`
	// For detail of what variables can be used, please check `RequestData`
	UrlTemplate string `comment:"GoTemplate for API url"`
	// Query would be sent out as part of the request URL
	Query func(reqData *RequestData) (url.Values, errors.Error)
	// Header would be sent out along with request
	Header func(reqData *RequestData) (http.Header, errors.Error)
	// GetTotalPages is to tell `ApiCollector` total number of pages based on response of the first page.
	// so `ApiCollector` could collect those pages in parallel for us
	GetTotalPages func(res *http.Response, args *ApiCollectorArgs) (int, errors.Error)
	// PageSize tells ApiCollector the page size
	PageSize int
	// GetNextPageCustomData indicate if this collection request each page in order and build query by the prev request
	GetNextPageCustomData func(prevReqData *RequestData, prevPageResponse *http.Response) (interface{}, errors.Error)
	// Incremental indicate if this is an incremental collection, the existing data won't get deleted if it was true
	Incremental bool `comment:"indicate if this collection is incremental update"`
	// ApiClient is a asynchronize api request client with qps
	ApiClient       RateLimitedApiClient
	MinTickInterval *time.Duration
	// Input helps us collect data based on previous collected data, like collecting changelogs based on jira
	// issue ids
	Input Iterator
	// Concurrency specify qps for api that doesn't return total number of pages/records
	// NORMALLY, DO NOT SPECIFY THIS PARAMETER, unless you know what it means
	Concurrency    int
	ResponseParser func(res *http.Response) ([]json.RawMessage, errors.Error)
	AfterResponse  plugin.ApiClientAfterResponse
	RequestBody    func(reqData *RequestData) map[string]interface{}
	Method         string
}

// ApiCollector FIXME ...
type ApiCollector struct {
	*RawDataSubTask
	args        *ApiCollectorArgs
	urlTemplate *template.Template
}

// NewApiCollector allocates a new ApiCollector with the given args.
// ApiCollector can help us to collect data from api with ease, pass in a AsyncApiClient and tell it which part
// of response we want to save, ApiCollector will collect them from remote server and store them into database.
func NewApiCollector(args ApiCollectorArgs) (*ApiCollector, errors.Error) {
	// process args
	rawDataSubTask, err := NewRawDataSubTask(args.RawDataSubTaskArgs)
	if err != nil {
		return nil, errors.Default.Wrap(err, "Couldn't resolve raw subtask args")
	}
	// TODO: check if args.Table is valid when this is a http GET request
	if args.UrlTemplate == "" && args.Method == "" {
		return nil, errors.Default.New("UrlTemplate is required")
	}
	tpl, err := errors.Convert01(template.New(args.Table).Parse(args.UrlTemplate))
	if err != nil {
		return nil, errors.Default.Wrap(err, "Failed to compile UrlTemplate")
	}
	if args.ApiClient == nil {
		return nil, errors.Default.New("ApiClient is required")
	}
	if args.ResponseParser == nil {
		return nil, errors.Default.New("ResponseParser is required")
	}
	apiCollector := &ApiCollector{
		RawDataSubTask: rawDataSubTask,
		args:           &args,
		urlTemplate:    tpl,
	}
	if args.AfterResponse != nil {
		apiCollector.SetAfterResponse(args.AfterResponse)
	} else {
		apiCollector.SetAfterResponse(func(res *http.Response) errors.Error {
			if res.StatusCode == http.StatusUnauthorized {
				return errors.Unauthorized.New("authentication failed, please check your AccessToken")
			}
			return nil
		})
	}
	return apiCollector, nil
}

var rawTableAutoMigrateLock sync.Mutex

func (collector *ApiCollector) ensureRawTable() errors.Error {
	db := collector.args.Ctx.GetDal()
	rawTableAutoMigrateLock.Lock()
	defer rawTableAutoMigrateLock.Unlock()
	return db.AutoMigrate(&RawData{}, dal.From(collector.table))
}

// Execute will start collection
func (collector *ApiCollector) Execute() errors.Error {
	logger := collector.args.Ctx.GetLogger()
	logger.Info("start api collection")

	// make sure table is created
	db := collector.args.Ctx.GetDal()
	err := collector.ensureRawTable()
	if err != nil {
		return errors.Default.Wrap(err, "error auto-migrating collector")
	}

	isIncremental := collector.args.Incremental
	syncPolicy := collector.args.Ctx.TaskContext().SyncPolicy()
	if syncPolicy != nil && syncPolicy.FullSync {
		isIncremental = false
	}
	// flush data if not incremental collection
	if !isIncremental {
		err = db.Delete(&RawData{}, dal.From(collector.table), dal.Where("params = ?", collector.params))
		if err != nil {
			return errors.Default.Wrap(err, "error deleting data from collector")
		}
	}

	// if MinTickInterval was specified
	if collector.args.MinTickInterval != nil {
		minTickInterval := *collector.args.MinTickInterval
		if minTickInterval <= time.Duration(0) {
			return errors.Default.Wrap(err, "MinTickInterval must be greater than 0")
		}
		oldTickInterval := collector.args.ApiClient.GetTickInterval()
		if oldTickInterval < minTickInterval {
			// reset the tick interval only if it exceeded the specified limit
			logger.Info("set tick interval to %v", minTickInterval.String())
			collector.args.ApiClient.Reset(minTickInterval)
			defer func() {
				logger.Info("restore tick interval to %v", oldTickInterval.String())
				collector.args.ApiClient.Reset(oldTickInterval)
			}()
		}
	}

	collector.args.Ctx.SetProgress(0, -1)
	if collector.args.Input != nil {
		iterator := collector.args.Input
		defer iterator.Close()
		apiClient := collector.args.ApiClient
		if apiClient == nil {
			return errors.Default.New("api_collector can not Execute with nil apiClient")
		}
		for {
			if !iterator.HasNext() || apiClient.HasError() {
				err = collector.args.ApiClient.WaitAsync()
				if err != nil {
					return err
				}
				if !iterator.HasNext() || apiClient.HasError() {
					break
				}
			}
			var input interface{}
			input, err = iterator.Fetch()
			if err != nil {
				break
			}
			collector.exec(input)
		}
	} else {
		// or we just did it once
		collector.exec(nil)
	}

	if err != nil {
		return errors.Default.Wrap(err, "error executing collector")
	}
	logger.Debug("wait for all async api to be finished")
	err = collector.args.ApiClient.WaitAsync()
	if err != nil {
		logger.Error(err, "end api collection error")
		err = errors.Default.Wrap(err, "Error waiting for async Collector execution")
	} else {
		logger.Info("end api collection without error")
	}

	return err
}

func (collector *ApiCollector) exec(input interface{}) {
	inputJson, err := json.Marshal(input)
	if err != nil {
		panic(err)
	}
	reqData := new(RequestData)
	reqData.Input = input
	reqData.InputJSON = inputJson
	reqData.Pager = &Pager{
		Page: 1,
		Size: collector.args.PageSize,
	}
	// fetch the detail
	if collector.args.PageSize <= 0 {
		collector.fetchAsync(reqData, nil)
		// fetch pages sequentially
	} else if collector.args.GetNextPageCustomData != nil {
		collector.fetchPagesSequentially(reqData)
		// fetch pages in parallel with number of total pages can be determined from the first page
	} else if collector.args.GetTotalPages != nil {
		collector.fetchPagesDetermined(reqData)
		// fetch pages in parallel without number of total pages
	} else {
		collector.fetchPagesUndetermined(reqData, false)
	}
}

// fetchPagesSequentially fetches data of all pages in order to build RequestData by prev response
func (collector *ApiCollector) fetchPagesSequentially(reqData *RequestData) {
	var collect func() errors.Error
	collect = func() errors.Error {
		collector.fetchAsync(reqData, func(count int, body []byte, res *http.Response) errors.Error {
			if count < collector.args.PageSize {
				return nil
			}
			customData, err := collector.args.GetNextPageCustomData(reqData, res)
			if err != nil {
				if errors.Is(err, ErrFinishCollect) {
					return nil
				} else {
					panic(err)
				}
			}
			reqData.CustomData = customData
			reqData.Pager.Skip += collector.args.PageSize
			reqData.Pager.Page += 1
			collector.args.ApiClient.NextTick(collect)
			return nil
		})
		return nil
	}
	collector.args.ApiClient.NextTick(collect)
}

// fetchPagesDetermined fetches data of all pages for APIs that return paging information
func (collector *ApiCollector) fetchPagesDetermined(reqData *RequestData) {
	// fetch first page
	collector.fetchAsync(reqData, func(count int, body []byte, res *http.Response) errors.Error {
		totalPages, err := collector.args.GetTotalPages(res, collector.args)
		if err != nil {
			// Some APIs might or might not return total pages/records based on total number of records
			// check https://github.com/apache/incubator-devlake/issues/8187 for details
			if err == ErrUndetermined {
				collector.fetchPagesUndetermined(reqData, true)
				return nil
			}
			return errors.Default.Wrap(err, "fetchPagesDetermined get totalPages failed")
		}
		// spawn a none blocking go routine to fetch other pages
		collector.args.ApiClient.NextTick(func() errors.Error {
			for page := 2; page <= totalPages; page++ {
				reqDataTemp := &RequestData{
					Pager: &Pager{
						Page: page,
						Skip: collector.args.PageSize * (page - 1),
						Size: collector.args.PageSize,
					},
					Input:     reqData.Input,
					InputJSON: reqData.InputJSON,
				}
				collector.fetchAsync(reqDataTemp, nil)
			}
			return nil
		})
		return nil
	})
}

// fetchPagesUndetermined fetches data of all pages for APIs that do NOT return paging information
func (collector *ApiCollector) fetchPagesUndetermined(reqData *RequestData, skipFirstPage bool) {
	//logger := collector.args.Ctx.GetLogger()
	//logger.Debug("fetch all pages in parallel with specified concurrency: %d", collector.args.Concurrency)
	// if api doesn't return total number of pages, employ a step concurrent technique
	// when `Concurrency` was set to 3:
	// goroutine #1 fetches pages 1/4/7..
	// goroutine #2 fetches pages 2/5/8...
	// goroutine #3 fetches pages 3/6/9...
	apiClient := collector.args.ApiClient
	concurrency := collector.args.Concurrency
	if concurrency == 0 {
		// normally when a multi-pages api depends on a another resource, like jira changelogs depend on issue ids
		// it tends to have less page, like 1 or 2 pages in total
		if collector.args.Input != nil {
			concurrency = 2
		} else {
			concurrency = apiClient.GetNumOfWorkers() / 10
			if concurrency < 10 {
				concurrency = 10
			}
		}
	}
	for i := 0; i < concurrency; i++ {
		reqDataCopy := RequestData{
			Pager: &Pager{
				Page: i + 1,
				Size: collector.args.PageSize,
				Skip: collector.args.PageSize * (i),
			},
			Input:     reqData.Input,
			InputJSON: reqData.InputJSON,
		}
		if skipFirstPage && reqDataCopy.Pager.Page == 1 {
			reqDataCopy.Pager.Page += concurrency
		}
		var collect func() errors.Error
		collect = func() errors.Error {
			collector.fetchAsync(&reqDataCopy, func(count int, body []byte, res *http.Response) errors.Error {
				if count < collector.args.PageSize {
					return nil
				}
				apiClient.NextTick(func() errors.Error {
					reqDataCopy.Pager.Skip += collector.args.PageSize * concurrency
					reqDataCopy.Pager.Page += concurrency
					return collect()
				})
				return nil
			})
			return nil
		}
		apiClient.NextTick(collect)
	}
}

func (collector *ApiCollector) generateUrl(pager *Pager, input interface{}) (string, errors.Error) {
	params := collector.args.Params
	if collector.args.Options != nil {
		params = collector.args.Options.GetParams()
	}
	var buf bytes.Buffer
	err := collector.urlTemplate.Execute(&buf, &RequestData{
		Pager:  pager,
		Params: params,
		Input:  input,
	})
	if err != nil {
		return "", errors.Convert(err)
	}
	return buf.String(), nil
}

// GetAfterResponse return apiClient's afterResponseFunction
func (collector *ApiCollector) GetAfterResponse() plugin.ApiClientAfterResponse {
	return collector.args.ApiClient.GetAfterFunction()
}

// SetAfterResponse set apiClient's afterResponseFunction
func (collector *ApiCollector) SetAfterResponse(f plugin.ApiClientAfterResponse) {
	collector.args.ApiClient.SetAfterFunction(f)
}

func (collector *ApiCollector) fetchAsync(reqData *RequestData, handler func(int, []byte, *http.Response) errors.Error) {
	if reqData.Pager == nil {
		reqData.Pager = &Pager{
			Page: 1,
			Size: 100,
			Skip: 0,
		}
	}
	apiUrl, err := collector.generateUrl(reqData.Pager, reqData.Input)
	if err != nil {
		panic(err)
	}
	var apiQuery url.Values
	if collector.args.Query != nil {
		apiQuery, err = collector.args.Query(reqData)
		if err != nil {
			panic(err)
		}
	}
	var reqBody interface{}
	if collector.args.RequestBody != nil {
		reqBody = collector.args.RequestBody(reqData)
		if err != nil {
			panic(err)
		}
	}

	apiHeader := (http.Header)(nil)
	if collector.args.Header != nil {
		apiHeader, err = collector.args.Header(reqData)
		if err != nil {
			panic(err)
		}
	}
	logger := collector.args.Ctx.GetLogger()
	logger.Debug("fetchAsync <<< enqueueing for %s %v", apiUrl, apiQuery)
	responseHandler := func(res *http.Response) errors.Error {
		defer logger.Debug("fetchAsync >>> done for %s %v", apiUrl, apiQuery)
		logger := collector.args.Ctx.GetLogger()
		// read body to buffer
		body, err := io.ReadAll(res.Body)
		if err != nil {
			return errors.Default.Wrap(err, fmt.Sprintf("error reading response from %s", apiUrl))
		}
		res.Body.Close()
		res.Body = io.NopCloser(bytes.NewBuffer(body))
		// convert body to array of RawJSON
		items, err := collector.args.ResponseParser(res)
		if err != nil {
			if errors.Is(err, ErrFinishCollect) {
				logger.Info("a fetch stop by parser, reqInput: #%s", reqData.Params)
				handler = nil
			} else {
				return errors.Default.Wrap(err, fmt.Sprintf("error parsing response from %s", apiUrl))
			}
		}
		// save to db
		count := len(items)
		if count == 0 {
			collector.args.Ctx.IncProgress(1)
			return nil
		}
		db := collector.args.Ctx.GetDal()
		urlString := res.Request.URL.String()
		rows := make([]*RawData, count)
		for i, msg := range items {
			rows[i] = &RawData{
				Params: collector.params,
				Data:   msg,
				Url:    urlString,
				Input:  reqData.InputJSON,
			}
		}
		err = db.Create(rows, dal.From(collector.table))
		if err != nil {
			return errors.Default.Wrap(err, fmt.Sprintf("error inserting raw rows into %s", collector.table))
		}
		logger.Debug("fetchAsync === total %d rows were saved into database", count)
		// increase progress only when it was not nested
		collector.args.Ctx.IncProgress(1)
		if handler != nil {
			// trigger next fetch, but return if ErrFinishCollect got from ResponseParser
			res.Body = io.NopCloser(bytes.NewBuffer(body))
			return handler(count, body, res)
		}
		return nil
	}
	if collector.args.Method == http.MethodPost {
		collector.args.ApiClient.DoPostAsync(apiUrl, apiQuery, reqBody, apiHeader, responseHandler)
	} else {
		collector.args.ApiClient.DoGetAsync(apiUrl, apiQuery, apiHeader, responseHandler)
	}
	logger.Debug("fetchAsync === enqueued for %s %v", apiUrl, apiQuery)
}