Repository: lf-edge/ekuiper
Pull request #2959 (2024-06-28 02:54:06)

fix(file): scanner dynamic buffer
Currently, if line is big, the lines scanner exit silently

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `61.90476%` with `8 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@484442f`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/file/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zb3VyY2UuZ28=) | 54.55% | [2 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/file/reader/lines.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Freader%2Flines.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9yZWFkZXIvbGluZXMuZ28=) | 66.67% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2959   +/-   ##
============================================
  Coverage             ?   69.74%           
============================================
  Files                ?      312           
  Lines                ?    29870           
  Branches             ?        0           
============================================
  Hits                 ?    20832           
  Misses               ?     7223           
  Partials             ?     1815           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2959?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/io/file/reader/csv.go b/internal/io/file/reader/csv.go
index 7cab72f3..fa1297aa 100644
--- a/internal/io/file/reader/csv.go
+++ b/internal/io/file/reader/csv.go
@@ -60,7 +60,7 @@ func (r *CsvReader) Provision(ctx api.StreamContext, props map[string]any) error
 	return nil
 }
 
-func (r *CsvReader) Bind(ctx api.StreamContext, fileStream io.Reader) error {
+func (r *CsvReader) Bind(ctx api.StreamContext, fileStream io.Reader, _ int) error {
 	cr := csv.NewReader(fileStream)
 	cr.Comma = rune(r.config.Delimiter[0])
 	cr.TrimLeadingSpace = true
diff --git a/internal/io/file/reader/lines.go b/internal/io/file/reader/lines.go
index 5c437cf8..74fcc7af 100644
--- a/internal/io/file/reader/lines.go
+++ b/internal/io/file/reader/lines.go
@@ -1,3 +1,17 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed 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 reader
 
 import (
@@ -22,8 +36,14 @@ func (r *LinesReader) Provision(ctx api.StreamContext, props map[string]any) err
 	return nil
 }
 
-func (r *LinesReader) Bind(ctx api.StreamContext, fileStream io.Reader) error {
+func (r *LinesReader) Bind(ctx api.StreamContext, fileStream io.Reader, maxSize int) error {
+	if maxSize <= 0 {
+		ctx.GetLogger().Errorf("maxSize must be > 0, defaul to 1MB")
+		// default to 1MB
+		maxSize = 1 << 20
+	}
 	scanner := bufio.NewScanner(fileStream)
+	scanner.Buffer(nil, maxSize)
 	scanner.Split(bufio.ScanLines)
 	r.scanner = scanner
 	return nil
@@ -34,7 +54,10 @@ func (r *LinesReader) Read(ctx api.StreamContext) (any, error) {
 	if !succ {
 		return nil, io.EOF
 	}
-	return r.scanner.Bytes(), nil
+	b := r.scanner.Bytes()
+	d := make([]byte, len(b))
+	copy(d, b)
+	return d, nil
 }
 
 func (r *LinesReader) IsBytesReader() bool {
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 4d88ae37..a313426f 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -214,11 +214,20 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 		err error
 		r   io.Reader
 	)
-	r, err = os.Open(file)
+	f, err := os.Open(file)
 	if err != nil {
 		ctx.GetLogger().Debugf("prepare file %s error: %v", file, err)
 		ingestError(ctx, err)
 	}
+	r = f
+	// This is the buffer size, 1MB by default
+	maxSize := 1 << 20
+	info, err := f.Stat()
+	if err != nil {
+		ctx.GetLogger().Debugf("get file info for %s error: %v", file, err)
+	} else {
+		maxSize = int(info.Size())
+	}
 	if fs.config.IgnoreStartLines > 0 || fs.config.IgnoreEndLines > 0 {
 		r = ignoreLines(ctx, r, fs.config.IgnoreStartLines, fs.config.IgnoreEndLines)
 	}
@@ -231,7 +240,7 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 	meta := map[string]any{"file": file}
 	// Read line or read all
 	if fs.reader != nil {
-		err = fs.reader.Bind(ctx, r)
+		err = fs.reader.Bind(ctx, r, maxSize)
 		if err != nil {
 			ingestError(ctx, err)
 			return
@@ -239,6 +248,9 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 		for {
 			line, err := fs.reader.Read(ctx)
 			if err != nil {
+				if err != io.EOF {
+					ctx.GetLogger().Errorf("read file %s error: %v", file, err)
+				}
 				break
 			}
 			rcvTime := timex.GetNow()
diff --git a/internal/io/http/client.go b/internal/io/http/client.go
index 9a1136d8..5288c8af 100644
--- a/internal/io/http/client.go
+++ b/internal/io/http/client.go
@@ -32,6 +32,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cert"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 // ClientConf is the configuration for http client
@@ -41,6 +42,25 @@ type ClientConf struct {
 	client       *http.Client
 	compressor   message.Compressor   // compressor used to payload compression when specifies compressAlgorithm
 	decompressor message.Decompressor // decompressor used to payload decompression when specifies compressAlgorithm
+
+	// auth related
+	accessConf        *AccessTokenConf
+	refreshConf       *RefreshTokenConf
+	tokenLastUpdateAt time.Time
+	tokens            map[string]interface{}
+}
+
+type AccessTokenConf struct {
+	Url            string `json:"url"`
+	Body           string `json:"body"`
+	Expire         string `json:"expire"`
+	ExpireInSecond int
+}
+
+type RefreshTokenConf struct {
+	Url     string            `json:"url"`
+	Headers map[string]string `json:"headers"`
+	Body    string            `json:"body"`
 }
 
 type RawConf struct {
@@ -50,6 +70,8 @@ type RawConf struct {
 	BodyType string            `json:"bodyType"`
 	Headers  map[string]string `json:"headers"`
 	Timeout  int               `json:"timeout"`
+
+	OAuth map[string]map[string]interface{} `json:"oauth"`
 	// Could be code or body
 	ResponseType string `json:"responseType"`
 	Compression  string `json:"compression"` // Compression specifies the algorithms used to payload compression
@@ -124,6 +146,39 @@ func (cc *ClientConf) InitConf(device string, props map[string]interface{}) erro
 	if err != nil {
 		return err
 	}
+	// validate oAuth. In order to adapt to manager, the validation is closed to allow empty value
+	if c.OAuth != nil {
+		// validate access token
+		if ap, ok := c.OAuth["access"]; ok {
+			accessConf := &AccessTokenConf{}
+			if err := cast.MapToStruct(ap, accessConf); err != nil {
+				return fmt.Errorf("fail to parse the access properties of oAuth: %v", err)
+			}
+			if accessConf.Url == "" {
+				conf.Log.Warnf("access token url is not set, so ignored the oauth setting")
+				c.OAuth = nil
+			} else {
+				// expire time will update every time when access token is refreshed if expired is set
+				cc.accessConf = accessConf
+			}
+		} else {
+			return fmt.Errorf("if setting oAuth, `access` property is required")
+		}
+		// validate refresh token, it is optional
+		if rp, ok := c.OAuth["refresh"]; ok {
+			refreshConf := &RefreshTokenConf{}
+			if err := cast.MapToStruct(rp, refreshConf); err != nil {
+				return fmt.Errorf("fail to parse the refresh token properties: %v", err)
+			}
+			if refreshConf.Url == "" {
+				conf.Log.Warnf("refresh token url is not set, so ignored the refresh setting")
+				delete(c.OAuth, "refresh")
+			} else {
+				cc.refreshConf = refreshConf
+			}
+		}
+	}
+
 	tlscfg, err := cert.GenTLSConfig(props, "http")
 	if err != nil {
 		return err
@@ -146,9 +201,92 @@ func (cc *ClientConf) InitConf(device string, props map[string]interface{}) erro
 			return fmt.Errorf("init payload decompressor failed, %w", err)
 		}
 	}
+	if cc.accessConf != nil {
+		conf.Log.Infof("Try to get access token from %s", cc.accessConf.Url)
+		ctx := mockContext.NewMockContext("none", "httppull_init")
+		cc.tokens = make(map[string]interface{})
+		err := cc.auth(ctx)
+		if err != nil {
+			return fmt.Errorf("fail to authorize by oAuth: %v", err)
+		}
+	}
+	return nil
+}
+
+// initialize the oAuth access token
+func (cc *ClientConf) auth(ctx api.StreamContext) error {
+	resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.accessConf.Url, nil, true, cc.accessConf.Body)
+	if err != nil {
+		return err
+	}
+	tokens, _, err := cc.parseResponse(ctx, resp)
+	if err != nil {
+		return err
+	}
+	cc.tokens = tokens[0]
+	ctx.GetLogger().Infof("Got access token %v", cc.tokens)
+	expireIn, err := ctx.ParseTemplate(cc.accessConf.Expire, cc.tokens)
+	if err != nil {
+		return fmt.Errorf("fail to parse the expire time for access token: %v", err)
+	}
+	cc.accessConf.ExpireInSecond, err = cast.ToInt(expireIn, cast.CONVERT_ALL)
+	if err != nil {
+		return fmt.Errorf("fail to covert the expire time %s for access token: %v", expireIn, err)
+	}
+	if cc.refreshConf != nil {
+		err := cc.refresh(ctx)
+		if err != nil {
+			return err
+		}
+	} else {
+		cc.tokenLastUpdateAt = time.Now()
+	}
 	return nil
 }
 
+func parseHeaders(ctx api.StreamContext, oHeaders map[string]string, data map[string]interface{}) (map[string]string, error) {
+	headers := make(map[string]string, len(oHeaders))
+	var err error
+	for k, v := range oHeaders {
+		headers[k], err = ctx.ParseTemplate(v, data)
+		if err != nil {
+			return nil, fmt.Errorf("fail to parse the header for refresh token request %s: %v", k, err)
+		}
+	}
+	return headers, nil
+}
+
+func (cc *ClientConf) refresh(ctx api.StreamContext) error {
+	if cc.refreshConf != nil {
+		headers := make(map[string]string, len(cc.refreshConf.Headers))
+		var err error
+		for k, v := range cc.refreshConf.Headers {
+			headers[k], err = ctx.ParseTemplate(v, cc.tokens)
+			if err != nil {
+				return fmt.Errorf("fail to parse the header for refresh token request %s: %v", k, err)
+			}
+		}
+
+		resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.refreshConf.Url, headers, true, cc.refreshConf.Body)
+		if err != nil {
+			return fmt.Errorf("fail to get refresh token: %v", err)
+		}
+		nt, _, err := cc.parseResponse(ctx, resp)
+		if err != nil {
+			return fmt.Errorf("Cannot parse refresh token response to json: %v", err)
+		}
+		for k, v := range nt[0] {
+			if v != nil {
+				cc.tokens[k] = v
+			}
+		}
+		cc.tokenLastUpdateAt = time.Now()
+		return nil
+	} else {
+		return fmt.Errorf("no oAuth config")
+	}
+}
+
 const (
 	BODY_ERR = "response body error"
 	CODE_ERR = "response code error"
@@ -224,6 +362,10 @@ func (cc *ClientConf) parseResponse(ctx api.StreamContext, resp *http.Response)
 	}
 }
 
+func (cc *ClientConf) parseHeaders(ctx api.StreamContext) (map[string]string, error) {
+	return parseHeaders(ctx, cc.config.Headers, cc.tokens)
+}
+
 func decode(data []byte) ([]map[string]interface{}, error) {
 	var r1 interface{}
 	err := json.Unmarshal(data, &r1)
diff --git a/internal/io/http/httppull_source.go b/internal/io/http/httppull_source.go
index 4224f1db..93b6b755 100644
--- a/internal/io/http/httppull_source.go
+++ b/internal/io/http/httppull_source.go
@@ -59,7 +59,11 @@ func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]a
 }
 
 func (hps *HttpPullSource) doPull(ctx api.StreamContext) ([]map[string]any, error) {
-	resp, err := httpx.Send(ctx.GetLogger(), hps.client, hps.config.BodyType, hps.config.Method, hps.config.Url, hps.config.Headers, true, []byte(hps.config.Body))
+	headers, err := hps.parseHeaders(ctx)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := httpx.Send(ctx.GetLogger(), hps.client, hps.config.BodyType, hps.config.Method, hps.config.Url, headers, true, []byte(hps.config.Body))
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/modules/file.go b/pkg/modules/file.go
index d8623c0b..880fdc48 100644
--- a/pkg/modules/file.go
+++ b/pkg/modules/file.go
@@ -47,7 +47,7 @@ type FileStreamReader interface {
 	// Provision Set up the static properties
 	Provision(ctx api.StreamContext, props map[string]any) error
 	// Bind set the file stream. Make sure the previous read has done
-	Bind(ctx api.StreamContext, fileStream io.Reader) error
+	Bind(ctx api.StreamContext, fileStream io.Reader, maxSize int) error
 	// Read the next record. Returns EOF when the input has reached its end.
 	Read(ctx api.StreamContext) (any, error)
 	// IsBytesReader If is bytes reader, Read must return []byte, otherwise return map or []map


Test patch:
diff --git a/internal/io/file/source_test.go b/internal/io/file/source_test.go
index a9e83de4..d5ad7112 100644
--- a/internal/io/file/source_test.go
+++ b/internal/io/file/source_test.go
@@ -184,7 +184,7 @@ func TestSourceProvision(t *testing.T) {
 			},
 		},
 		{
-			name: "no decompression for stream typs",
+			name: "no decompression for stream types",
 			props: map[string]any{
 				"datasource":    name,
 				"path":          path,
diff --git a/internal/io/http/client_test.go b/internal/io/http/client_test.go
index 7589024d..1e36d4c6 100644
--- a/internal/io/http/client_test.go
+++ b/internal/io/http/client_test.go
@@ -16,9 +16,12 @@ package http
 
 import (
 	"encoding/json"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestInitConf(t *testing.T) {
@@ -134,3 +137,31 @@ func TestDecode(t *testing.T) {
 		require.Equal(t, tc.got, g)
 	}
 }
+
+func TestClientAuth(t *testing.T) {
+	server := createServer()
+	defer func() {
+		server.Close()
+	}()
+
+	c := &ClientConf{}
+	ctx := mockContext.NewMockContext("1", "2")
+	require.NoError(t, c.InitConf("", map[string]interface{}{
+		"oauth": map[string]interface{}{
+			"access": map[string]interface{}{
+				"url":    fmt.Sprintf("%s/auth", server.URL),
+				"expire": "3600",
+				"body":   `{"a":1}`,
+			},
+			"refresh": map[string]interface{}{
+				"url": fmt.Sprintf("%s/refresh", server.URL),
+				"headers": map[string]interface{}{
+					"a": "{{.message}}",
+				},
+				"body": `{"a":1}`,
+			},
+		},
+	}))
+	require.NoError(t, c.auth(ctx))
+	require.NoError(t, c.refresh(ctx))
+}
diff --git a/internal/io/http/httppull_source_test.go b/internal/io/http/httppull_source_test.go
index 12020d23..e76cedaa 100644
--- a/internal/io/http/httppull_source_test.go
+++ b/internal/io/http/httppull_source_test.go
@@ -61,11 +61,33 @@ func handleErr(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(resp)
 }
 
+func handleAuth(w http.ResponseWriter, r *http.Request) {
+	resp := Response{
+		Message: "auth",
+		Code:    200,
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(resp)
+}
+
+func handleRefresh(w http.ResponseWriter, r *http.Request) {
+	resp := Response{
+		Message: "refresh",
+		Code:    200,
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(resp)
+}
+
 func createServer() *httptest.Server {
 	router := http.NewServeMux()
 	router.HandleFunc("/get", handleGet)
 	router.HandleFunc("/post", handlePost)
 	router.HandleFunc("/err", handleErr)
+	router.HandleFunc("/auth", handleAuth)
+	router.HandleFunc("/refresh", handleRefresh)
 	server := httptest.NewServer(router)
 	return server
 }