refactor: move from io/ioutil to io and os packages

The io/ioutil package has been deprecated as of Go 1.16 [1]. This commit
replaces the existing io/ioutil functions with their new definitions in
io and os packages.

[1]: https://golang.org/doc/go1.16#ioutil
Signed-off-by: Eng Zer Jun <engzerjun@gmail.com>

cmd/kuiper/main.go

@@ -17,16 +17,16 @@ package main
 import (
 	"bufio"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/model"
-	"github.com/lf-edge/ekuiper/pkg/infra"
-	"github.com/urfave/cli"
-	"io/ioutil"
 	"net/rpc"
 	"os"
 	"sort"
 	"strings"
 	"time"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/pkg/infra"
+	"github.com/urfave/cli"
 )
 
 type clientConf struct {
@@ -1005,7 +1005,7 @@ func readDef(sfile string, t string) ([]byte, error) {
 		return nil, fmt.Errorf("The specified %s defenition file %s is not existed.\n", t, sfile)
 	}
 	fmt.Printf("Creating a new %s from file %s.\n", t, sfile)
-	if rule, err := ioutil.ReadFile(sfile); err != nil {
+	if rule, err := os.ReadFile(sfile); err != nil {
 		return nil, fmt.Errorf("Failed to read from %s definition file %s.\n", t, sfile)
 	} else {
 		return rule, nil

extensions/sinks/image/image.go

@@ -18,14 +18,14 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"github.com/lf-edge/ekuiper/pkg/api"
 	"image/jpeg"
 	"image/png"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
+
+	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
 type imageSink struct {
@@ -102,11 +102,20 @@ func (m *imageSink) Open(ctx api.StreamContext) error {
 }
 
 func (m *imageSink) delFile(logger api.Logger) error {
-	files, err := ioutil.ReadDir(m.path)
-	if nil != err || 0 == len(files) {
+	dirEntries, err := os.ReadDir(m.path)
+	if nil != err || 0 == len(dirEntries) {
 		return err
 	}
 
+	files := make([]os.FileInfo, 0, len(dirEntries))
+	for _, entry := range dirEntries {
+		info, err := entry.Info()
+		if err != nil {
+			continue
+		}
+		files = append(files, info)
+	}
+
 	pos := m.maxCount
 	delTime := time.Now().Add(time.Duration(0-m.maxAge) * time.Hour)
 	for i := 0; i < len(files); i++ {

internal/conf/load.go

@@ -32,14 +32,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/mitchellh/mapstructure"
-	"gopkg.in/yaml.v3"
-	"io/ioutil"
 	"os"
 	"path"
 	"path/filepath"
 	"strconv"
 	"strings"
+
+	"github.com/mitchellh/mapstructure"
+	"gopkg.in/yaml.v3"
 )
 
 const Separator = "__"
@@ -59,7 +59,7 @@ func LoadConfigByName(name string, c interface{}) error {
 
 func LoadConfigFromPath(p string, c interface{}) error {
 	prefix := getPrefix(p)
-	b, err := ioutil.ReadFile(p)
+	b, err := os.ReadFile(p)
 	if err != nil {
 		return err
 	}
@@ -240,7 +240,7 @@ func extractKeysFromJsonIfExists(yamlPath string) ([]string, error) {
 }
 
 func loadJsonForYaml(filePath string) (map[string]interface{}, error) {
-	data, err := ioutil.ReadFile(filePath)
+	data, err := os.ReadFile(filePath)
 	if err != nil {
 		return nil, err
 	}

internal/converter/binary/converter_test.go

@@ -15,14 +15,14 @@
 package binary
 
 import (
-	"io/ioutil"
+	"os"
 	"path"
 	"reflect"
 	"testing"
 )
 
 func TestMessageDecode(t *testing.T) {
-	image, err := ioutil.ReadFile(path.Join("../../../docs", "cover.jpg"))
+	image, err := os.ReadFile(path.Join("../../../docs", "cover.jpg"))
 	if err != nil {
 		t.Errorf("Cannot read image: %v", err)
 	}

internal/converter/json/converter_test.go

@@ -17,14 +17,14 @@ package json
 import (
 	"encoding/base64"
 	"fmt"
-	"io/ioutil"
+	"os"
 	"path"
 	"reflect"
 	"testing"
 )
 
 func TestMessageDecode(t *testing.T) {
-	image, err := ioutil.ReadFile(path.Join("../../../docs", "cover.jpg"))
+	image, err := os.ReadFile(path.Join("../../../docs", "cover.jpg"))
 	if err != nil {
 		t.Errorf("Cannot read image: %v", err)
 	}

internal/meta/func_meta.go

@@ -16,11 +16,12 @@ package meta
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/filex"
-	"io/ioutil"
+	"os"
 	"path"
 	"strings"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 )
 
 type (
@@ -89,7 +90,7 @@ func ReadFuncMetaDir(checker InstallChecker) error {
 	}
 
 	dir := path.Join(confDir, "functions")
-	files, err := ioutil.ReadDir(dir)
+	files, err := os.ReadDir(dir)
 	if nil != err {
 		return err
 	}

internal/meta/msgUtil.go

@@ -15,10 +15,11 @@
 package meta
 
 import (
+	"os"
+	"path"
+
 	kconf "github.com/lf-edge/ekuiper/internal/conf"
 	"gopkg.in/ini.v1"
-	"io/ioutil"
-	"path"
 )
 
 var gUimsg map[string]*ini.File
@@ -41,13 +42,13 @@ func ReadUiMsgDir() error {
 	}
 
 	dir := path.Join(confDir, "multilingual")
-	infos, err := ioutil.ReadDir(dir)
+	dirEntries, err := os.ReadDir(dir)
 	if nil != err {
 		return err
 	}
 
-	for _, info := range infos {
-		fName := info.Name()
+	for _, entry := range dirEntries {
+		fName := entry.Name()
 		kconf.Log.Infof("uiMsg file : %s", fName)
 		fPath := path.Join(dir, fName)
 		if conf, err := ini.Load(fPath); nil != err {

internal/meta/sinkMeta.go

@@ -16,12 +16,13 @@ package meta
 
 import (
 	"fmt"
+	"os"
+	"path"
+	"strings"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 	"github.com/lf-edge/ekuiper/pkg/cast"
-	"io/ioutil"
-	"path"
-	"strings"
 )
 
 const (
@@ -189,7 +190,7 @@ func ReadSinkMetaDir(checker InstallChecker) error {
 	}
 
 	dir := path.Join(confDir, "sinks")
-	files, err := ioutil.ReadDir(dir)
+	files, err := os.ReadDir(dir)
 	if nil != err {
 		return err
 	}

internal/meta/sourceMeta.go

@@ -16,12 +16,13 @@ package meta
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/filex"
-	"io/ioutil"
+	"os"
 	"path"
 	"strings"
 	"sync"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 )
 
 type (
@@ -111,7 +112,7 @@ func ReadSourceMetaDir(checker InstallChecker) error {
 	}
 
 	dir := path.Join(confDir, "sources")
-	infos, err := ioutil.ReadDir(dir)
+	dirEntries, err := os.ReadDir(dir)
 	if nil != err {
 		return err
 	}
@@ -121,8 +122,8 @@ func ReadSourceMetaDir(checker InstallChecker) error {
 	}
 	conf.Log.Infof("Loading metadata file for source : %s", "mqtt_source.json")
 
-	for _, info := range infos {
-		fileName := info.Name()
+	for _, entry := range dirEntries {
+		fileName := entry.Name()
 		if strings.HasSuffix(fileName, ".json") {
 			filePath := path.Join(dir, fileName)
 			if err = ReadSourceMetaFile(filePath, checker(strings.TrimSuffix(fileName, ".json"))); nil != err {

internal/pkg/cert/cert.go

@@ -3,8 +3,9 @@ package cert
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"os"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"io/ioutil"
 )
 
 type TlsConfigurationOptions struct {
@@ -61,7 +62,7 @@ func certLoader(certFilePath, keyFilePath string) (tls.Certificate, error) {
 func caLoader(caFilePath string) (*x509.CertPool, error) {
 	if cp, err := conf.ProcessPath(caFilePath); err == nil {
 		pool := x509.NewCertPool()
-		caCrt, err1 := ioutil.ReadFile(cp)
+		caCrt, err1 := os.ReadFile(cp)
 		if err1 != nil {
 			return nil, err1
 		}

internal/pkg/filex/parser.go

@@ -16,12 +16,13 @@ package filex
 
 import (
 	"encoding/json"
+	"os"
+
 	"gopkg.in/yaml.v3"
-	"io/ioutil"
 )
 
 func ReadJsonUnmarshal(path string, ret interface{}) error {
-	sliByte, err := ioutil.ReadFile(path)
+	sliByte, err := os.ReadFile(path)
 	if nil != err {
 		return err
 	}
@@ -36,11 +37,11 @@ func WriteYamlMarshal(path string, data interface{}) error {
 	if nil != err {
 		return err
 	}
-	return ioutil.WriteFile(path, y, 0666)
+	return os.WriteFile(path, y, 0666)
 }
 
 func ReadYamlUnmarshal(path string, ret interface{}) error {
-	sliByte, err := ioutil.ReadFile(path)
+	sliByte, err := os.ReadFile(path)
 	if nil != err {
 		return err
 	}

internal/pkg/httpx/http.go

@@ -19,15 +19,15 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/pkg/api"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"net/url"
 	"os"
 	"strings"
 	"time"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
 var BodyTypeMap = map[string]string{"none": "", "text": "text/plain", "json": "application/json", "html": "text/html", "xml": "application/xml", "javascript": "application/javascript", "form": ""}
@@ -79,7 +79,7 @@ func Send(logger api.Logger, client *http.Client, bodyType string, method string
 			}
 			form.Set(key, vstr)
 		}
-		body := ioutil.NopCloser(strings.NewReader(form.Encode()))
+		body := io.NopCloser(strings.NewReader(form.Encode()))
 		req, err = http.NewRequest(method, u, body)
 		if err != nil {
 			return nil, fmt.Errorf("fail to create request: %v", err)

internal/pkg/jwt/rsa_keys.go

@@ -17,12 +17,13 @@ package jwt
 import (
 	"crypto/rsa"
 	"fmt"
-	"github.com/golang-jwt/jwt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"io/ioutil"
+	"os"
 	"path"
 	"strings"
 	"sync"
+
+	"github.com/golang-jwt/jwt"
+	"github.com/lf-edge/ekuiper/internal/conf"
 )
 
 var privateKeyRepository = make(map[string]*rsa.PrivateKey)
@@ -65,13 +66,13 @@ func insensitiveGetFilePath(prikeyName string) (string, error) {
 	}
 
 	dir := path.Join(confDir, RSAKeyDir)
-	infos, err := ioutil.ReadDir(dir)
+	dirEntries, err := os.ReadDir(dir)
 	if nil != err {
 		return "", err
 	}
 
-	for _, info := range infos {
-		fileName := info.Name()
+	for _, entry := range dirEntries {
+		fileName := entry.Name()
 		if strings.EqualFold(fileName, prikeyName) {
 			filePath := path.Join(dir, fileName)
 			return filePath, nil
@@ -85,7 +86,7 @@ func privateKeyFromFile(keyName string) (*rsa.PrivateKey, error) {
 	if err != nil {
 		return nil, err
 	}
-	keyBytes, err := ioutil.ReadFile(keyPath)
+	keyBytes, err := os.ReadFile(keyPath)
 	if err != nil {
 		return nil, err
 	}
@@ -101,7 +102,7 @@ func publicKeyFromFile(keyName string) (*rsa.PublicKey, error) {
 	if err != nil {
 		return nil, err
 	}
-	keyBytes, err := ioutil.ReadFile(keyPath)
+	keyBytes, err := os.ReadFile(keyPath)
 	if err != nil {
 		return nil, err
 	}

internal/plugin/native/manager.go

@@ -20,16 +20,6 @@ import (
 	"archive/zip"
 	"bytes"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/meta"
-	"github.com/lf-edge/ekuiper/internal/pkg/filex"
-	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
-	"github.com/lf-edge/ekuiper/internal/pkg/store"
-	plugin2 "github.com/lf-edge/ekuiper/internal/plugin"
-	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"github.com/lf-edge/ekuiper/pkg/kv"
-	"io/ioutil"
 	"os"
 	"os/exec"
 	"path"
@@ -40,6 +30,16 @@ import (
 	"sync"
 	"time"
 	"unicode"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/meta"
+	"github.com/lf-edge/ekuiper/internal/pkg/filex"
+	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
+	plugin2 "github.com/lf-edge/ekuiper/internal/plugin"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"github.com/lf-edge/ekuiper/pkg/kv"
 )
 
 // Manager Initialized in the binder
@@ -47,7 +47,7 @@ var manager *Manager
 
 const DELETED = "$deleted"
 
-//Manager is append only because plugin cannot delete or reload. To delete a plugin, restart the server to reindex
+// Manager is append only because plugin cannot delete or reload. To delete a plugin, restart the server to reindex
 type Manager struct {
 	sync.RWMutex
 	// 3 maps for source/sink/function. In each map, key is the plugin name, value is the version
@@ -105,7 +105,7 @@ func InitManager() (*Manager, error) {
 func findAll(t plugin2.PluginType, pluginDir string) (result map[string]string, err error) {
 	result = make(map[string]string)
 	dir := path.Join(pluginDir, plugin2.PluginTypes[t])
-	files, err := ioutil.ReadDir(dir)
+	files, err := os.ReadDir(dir)
 	if err != nil {
 		return
 	}

internal/plugin/portable/manager.go

@@ -19,19 +19,20 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/meta"
-	"github.com/lf-edge/ekuiper/internal/pkg/filex"
-	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
-	"github.com/lf-edge/ekuiper/internal/plugin"
-	"github.com/lf-edge/ekuiper/internal/plugin/portable/runtime"
-	"io/ioutil"
+	"io"
 	"os"
 	"os/exec"
 	"path"
 	"path/filepath"
 	"strings"
 	"sync"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/meta"
+	"github.com/lf-edge/ekuiper/internal/pkg/filex"
+	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
+	"github.com/lf-edge/ekuiper/internal/plugin"
+	"github.com/lf-edge/ekuiper/internal/plugin/portable/runtime"
 )
 
 var manager *Manager
@@ -97,7 +98,7 @@ func MockManager(plugins map[string]*PluginInfo) (*Manager, error) {
 }
 
 func (m *Manager) syncRegistry() error {
-	files, err := ioutil.ReadDir(m.pluginDir)
+	files, err := os.ReadDir(m.pluginDir)
 	if err != nil {
 		return fmt.Errorf("read path '%s' error: %v", m.pluginDir, err)
 	}
@@ -230,7 +231,7 @@ func (m *Manager) install(name, src string, shellParas []string) (resultErr erro
 				return err
 			}
 			pi = &PluginInfo{PluginMeta: runtime.PluginMeta{Name: name}}
-			allBytes, err := ioutil.ReadAll(jf)
+			allBytes, err := io.ReadAll(jf)
 			if err != nil {
 				return err
 			}

internal/schema/registry.go

@@ -16,14 +16,14 @@ package schema
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/def"
-	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
 	"sync"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
 )
 
 // Initialize in the server startup
@@ -52,7 +52,7 @@ func InitRegistry() error {
 	for _, schemaType := range def.SchemaTypes {
 		schemaDir := filepath.Join(etcDir, "schemas", string(schemaType))
 		var newSchemas map[string]string
-		files, err := ioutil.ReadDir(schemaDir)
+		files, err := os.ReadDir(schemaDir)
 		if err != nil {
 			conf.Log.Warnf("cannot read schema directory: %s", err)
 			newSchemas = make(map[string]string)
@@ -130,7 +130,7 @@ func GetSchema(schemaType def.SchemaType, name string) (*Info, error) {
 	if err != nil {
 		return nil, err
 	}
-	content, err := ioutil.ReadFile(schemaFile)
+	content, err := os.ReadFile(schemaFile)
 	if err != nil {
 		return nil, fmt.Errorf("cannot read schema file %s: %s", schemaFile, err)
 	}

internal/schema/registry_test.go

@@ -15,15 +15,15 @@
 package schema
 
 import (
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/testx"
-	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"reflect"
 	"testing"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/testx"
 )
 
 func TestRegistry(t *testing.T) {
@@ -39,11 +39,11 @@ func TestRegistry(t *testing.T) {
 		t.Fatal(err)
 	}
 	//Copy init.proto
-	bytesRead, err := ioutil.ReadFile("test/init.proto")
+	bytesRead, err := os.ReadFile("test/init.proto")
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = ioutil.WriteFile(filepath.Join(etcDir, "init.proto"), bytesRead, 0755)
+	err = os.WriteFile(filepath.Join(etcDir, "init.proto"), bytesRead, 0755)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -164,7 +164,7 @@ func TestRegistry(t *testing.T) {
 }
 
 func checkFile(etcDir string, schemas []string, t *testing.T) {
-	files, err := ioutil.ReadDir(etcDir)
+	files, err := os.ReadDir(etcDir)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/server/meta_init.go

@@ -19,11 +19,12 @@ package server
 
 import (
 	"fmt"
-	"github.com/gorilla/mux"
-	"github.com/lf-edge/ekuiper/internal/meta"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"strings"
+
+	"github.com/gorilla/mux"
+	"github.com/lf-edge/ekuiper/internal/meta"
 )
 
 func init() {
@@ -58,7 +59,7 @@ func (m metaComp) rest(r *mux.Router) {
 	}
 }
 
-//list sink plugin
+// list sink plugin
 func sinksMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	sinks := meta.GetSinks()
@@ -66,7 +67,7 @@ func sinksMetaHandler(w http.ResponseWriter, r *http.Request) {
 	return
 }
 
-//Get sink metadata when creating rules
+// Get sink metadata when creating rules
 func newSinkMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -81,7 +82,7 @@ func newSinkMetaHandler(w http.ResponseWriter, r *http.Request) {
 	jsonResponse(ptrMetadata, w, logger)
 }
 
-//list functions
+// list functions
 func functionsMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	sinks := meta.GetFunctions()
@@ -89,7 +90,7 @@ func functionsMetaHandler(w http.ResponseWriter, r *http.Request) {
 	return
 }
 
-//list source plugin
+// list source plugin
 func sourcesMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	ret := meta.GetSourcesPlugins()
@@ -99,7 +100,7 @@ func sourcesMetaHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//list shareMeta
+// list shareMeta
 func connectionsMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	ret := meta.GetConnectionPlugins()
@@ -109,7 +110,7 @@ func connectionsMetaHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Get source metadata when creating stream
+// Get source metadata when creating stream
 func sourceMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -126,7 +127,7 @@ func sourceMetaHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Get source metadata when creating stream
+// Get source metadata when creating stream
 func connectionMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -143,7 +144,7 @@ func connectionMetaHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Get source yaml
+// Get source yaml
 func sourceConfHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -159,7 +160,7 @@ func sourceConfHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Get share yaml
+// Get share yaml
 func connectionConfHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -175,7 +176,7 @@ func connectionConfHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Add  del confkey
+// Add  del confkey
 func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 
 	defer r.Body.Close()
@@ -188,7 +189,7 @@ func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	case http.MethodDelete:
 		err = meta.DelSourceConfKey(pluginName, confKey, language)
 	case http.MethodPut:
-		v, err1 := ioutil.ReadAll(r.Body)
+		v, err1 := io.ReadAll(r.Body)
 		if err1 != nil {
 			handleError(w, err, "Invalid body", logger)
 			return
@@ -201,7 +202,7 @@ func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Add  del confkey
+// Add  del confkey
 func connectionConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 
 	defer r.Body.Close()
@@ -214,7 +215,7 @@ func connectionConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	case http.MethodDelete:
 		err = meta.DelConnectionConfKey(pluginName, confKey, language)
 	case http.MethodPut:
-		v, err1 := ioutil.ReadAll(r.Body)
+		v, err1 := io.ReadAll(r.Body)
 		if err1 != nil {
 			handleError(w, err1, "Invalid body", logger)
 			return

internal/server/middleware/auth_test.go

@@ -1,12 +1,13 @@
 package middleware
 
 import (
-	"github.com/lf-edge/ekuiper/internal/pkg/jwt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"reflect"
 	"testing"
+
+	"github.com/lf-edge/ekuiper/internal/pkg/jwt"
 )
 
 func genToken(signKeyName, issuer, aud string) string {
@@ -69,7 +70,7 @@ func Test_AUTH(t *testing.T) {
 
 			res := tt.res.Result()
 
-			data, err := ioutil.ReadAll(res.Body)
+			data, err := io.ReadAll(res.Body)
 			if err != nil {
 				t.Errorf("expected error to be nil got %v", err)
 			}

internal/server/rest.go

@@ -17,16 +17,7 @@ package server
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/gorilla/handlers"
-	"github.com/gorilla/mux"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/server/middleware"
-	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/ast"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"github.com/lf-edge/ekuiper/pkg/infra"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -34,6 +25,15 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/gorilla/handlers"
+	"github.com/gorilla/mux"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/server/middleware"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
 const (
@@ -221,7 +221,7 @@ func fileUploadHandler(w http.ResponseWriter, r *http.Request) {
 
 	case http.MethodGet:
 		// Get the list of files in the upload directory
-		files, err := ioutil.ReadDir(uploadDir)
+		files, err := os.ReadDir(uploadDir)
 		if err != nil {
 			handleError(w, err, "Error reading the file upload dir", logger)
 			return
@@ -254,7 +254,7 @@ type information struct {
 	UpTimeSeconds int64  `json:"upTimeSeconds"`
 }
 
-//The handler for root
+// The handler for root
 func rootHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	switch r.Method {
@@ -337,17 +337,17 @@ func sourceManageHandler(w http.ResponseWriter, r *http.Request, st ast.StreamTy
 	}
 }
 
-//list or create streams
+// list or create streams
 func streamsHandler(w http.ResponseWriter, r *http.Request) {
 	sourcesManageHandler(w, r, ast.TypeStream)
 }
 
-//describe or delete a stream
+// describe or delete a stream
 func streamHandler(w http.ResponseWriter, r *http.Request) {
 	sourceManageHandler(w, r, ast.TypeStream)
 }
 
-//list or create tables
+// list or create tables
 func tablesHandler(w http.ResponseWriter, r *http.Request) {
 	sourcesManageHandler(w, r, ast.TypeTable)
 }
@@ -356,12 +356,12 @@ func tableHandler(w http.ResponseWriter, r *http.Request) {
 	sourceManageHandler(w, r, ast.TypeTable)
 }
 
-//list or create rules
+// list or create rules
 func rulesHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	switch r.Method {
 	case http.MethodPost:
-		body, err := ioutil.ReadAll(r.Body)
+		body, err := io.ReadAll(r.Body)
 		if err != nil {
 			handleError(w, err, "Invalid body", logger)
 			return
@@ -401,7 +401,7 @@ func rulesHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//describe or delete a rule
+// describe or delete a rule
 func ruleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -432,7 +432,7 @@ func ruleHandler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		body, err := ioutil.ReadAll(r.Body)
+		body, err := io.ReadAll(r.Body)
 		if err != nil {
 			handleError(w, err, "Invalid body", logger)
 			return
@@ -457,7 +457,7 @@ func ruleHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//get status of a rule
+// get status of a rule
 func getStatusRuleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -473,7 +473,7 @@ func getStatusRuleHandler(w http.ResponseWriter, r *http.Request) {
 	w.Write([]byte(content))
 }
 
-//start a rule
+// start a rule
 func startRuleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -488,7 +488,7 @@ func startRuleHandler(w http.ResponseWriter, r *http.Request) {
 	w.Write([]byte(fmt.Sprintf("Rule %s was started", name)))
 }
 
-//stop a rule
+// stop a rule
 func stopRuleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -499,7 +499,7 @@ func stopRuleHandler(w http.ResponseWriter, r *http.Request) {
 	w.Write([]byte(result))
 }
 
-//restart a rule
+// restart a rule
 func restartRuleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
@@ -514,7 +514,7 @@ func restartRuleHandler(w http.ResponseWriter, r *http.Request) {
 	w.Write([]byte(fmt.Sprintf("Rule %s was restarted", name)))
 }
 
-//get topo of a rule
+// get topo of a rule
 func getTopoRuleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)

internal/service/executors.go

@@ -18,6 +18,16 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/rpc"
+	"net/url"
+	"reflect"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/golang/protobuf/proto"
 	"github.com/jhump/protoreflect/dynamic"
 	"github.com/jhump/protoreflect/dynamic/grpcdynamic"
@@ -27,15 +37,6 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/infra"
 	"github.com/ugorji/go/codec"
 	"google.golang.org/grpc"
-	"io/ioutil"
-	"net"
-	"net/http"
-	"net/rpc"
-	"net/url"
-	"reflect"
-	"strings"
-	"sync"
-	"time"
 )
 
 // NewExecutor
@@ -215,11 +216,11 @@ func (h *httpExecutor) InvokeFunction(ctx api.FunctionContext, name string, para
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
-		buf, _ := ioutil.ReadAll(resp.Body)
+		buf, _ := io.ReadAll(resp.Body)
 		ctx.GetLogger().Debugf("%s\n", string(buf))
 		return nil, fmt.Errorf("http executor fails to err http return code: %d and error message %s", resp.StatusCode, string(buf))
 	} else {
-		buf, bodyErr := ioutil.ReadAll(resp.Body)
+		buf, bodyErr := io.ReadAll(resp.Body)
 		if bodyErr != nil {
 			return nil, fmt.Errorf("http executor read response body error: %v", bodyErr)
 		}

internal/service/manager.go

@@ -17,18 +17,18 @@ package service
 import (
 	"archive/zip"
 	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"sync"
+
 	kconf "github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
 	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/kv"
-	"io/ioutil"
-	"os"
-	"path"
-	"path/filepath"
-	"strings"
-	"sync"
 )
 
 var (
@@ -99,7 +99,7 @@ func GetManager() *Manager {
  */
 func (m *Manager) InitByFiles() error {
 	kconf.Log.Debugf("init service manager")
-	files, err := ioutil.ReadDir(m.etcDir)
+	files, err := os.ReadDir(m.etcDir)
 	if nil != err {
 		return err
 	}

internal/topo/node/sink_node_test.go

@@ -20,18 +20,18 @@ package node
 import (
 	"errors"
 	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+	"time"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/schema"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/topotest/mocknode"
 	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/internal/xsql"
-	"io/ioutil"
-	"os"
-	"path/filepath"
-	"reflect"
-	"testing"
-	"time"
 )
 
 func TestSinkTemplate_Apply(t *testing.T) {
@@ -240,11 +240,11 @@ func TestFormat_Apply(t *testing.T) {
 		t.Fatal(err)
 	}
 	//Copy init.proto
-	bytesRead, err := ioutil.ReadFile("../../schema/test/test1.proto")
+	bytesRead, err := os.ReadFile("../../schema/test/test1.proto")
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = ioutil.WriteFile(filepath.Join(etcDir, "test1.proto"), bytesRead, 0755)
+	err = os.WriteFile(filepath.Join(etcDir, "test1.proto"), bytesRead, 0755)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/topo/operator/preprocessor_test.go

@@ -19,6 +19,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
+	"os"
+	"path"
+	"reflect"
+	"testing"
+	"time"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/converter"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
@@ -26,12 +33,6 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/message"
-	"io/ioutil"
-	"log"
-	"path"
-	"reflect"
-	"testing"
-	"time"
 )
 
 func TestPreprocessor_Apply(t *testing.T) {
@@ -1003,7 +1004,7 @@ func TestPreprocessorForBinary(t *testing.T) {
 	if err != nil {
 		t.Errorf("Cannot find docs folder: %v", err)
 	}
-	image, err := ioutil.ReadFile(path.Join(docsFolder, "cover.jpg"))
+	image, err := os.ReadFile(path.Join(docsFolder, "cover.jpg"))
 	if err != nil {
 		t.Errorf("Cannot read image: %v", err)
 	}

internal/topo/sink/rest_sink.go

@@ -17,15 +17,16 @@ package sink
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/pkg/cert"
-	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
-	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/url"
 	"strings"
 	"time"
+
+	"github.com/lf-edge/ekuiper/internal/pkg/cert"
+	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
 )
 
 type RestSink struct {
@@ -240,7 +241,7 @@ func (ms *RestSink) Collect(ctx api.StreamContext, item interface{}) error {
 		defer resp.Body.Close()
 		logger.Debugf("rest sink got response %v", resp)
 		if resp.StatusCode < 200 || resp.StatusCode > 299 {
-			if buf, bodyErr := ioutil.ReadAll(resp.Body); bodyErr != nil {
+			if buf, bodyErr := io.ReadAll(resp.Body); bodyErr != nil {
 				logger.Errorf("%s\n", bodyErr)
 				return fmt.Errorf("%s: http return code: %d and error message %s", errorx.IOErr, resp.StatusCode, bodyErr)
 			} else {
@@ -249,7 +250,7 @@ func (ms *RestSink) Collect(ctx api.StreamContext, item interface{}) error {
 			}
 		} else {
 			if ms.debugResp {
-				if buf, bodyErr := ioutil.ReadAll(resp.Body); bodyErr != nil {
+				if buf, bodyErr := io.ReadAll(resp.Body); bodyErr != nil {
 					logger.Errorf("%s\n", bodyErr)
 				} else {
 					logger.Infof("Response content: %s\n", string(buf))

internal/topo/sink/rest_sink_test.go

@@ -16,14 +16,15 @@ package sink
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/transform"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"reflect"
 	"testing"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/transform"
 )
 
 type request struct {
@@ -170,7 +171,7 @@ func TestRestSink_Apply(t *testing.T) {
 
 	var requests []request
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, err := ioutil.ReadAll(r.Body)
+		body, err := io.ReadAll(r.Body)
 		if err != nil {
 			fmt.Printf("Error reading body: %v", err)
 			http.Error(w, "can't read body", http.StatusBadRequest)
@@ -330,7 +331,7 @@ func TestRestSinkTemplate_Apply(t *testing.T) {
 
 	var requests []request
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, err := ioutil.ReadAll(r.Body)
+		body, err := io.ReadAll(r.Body)
 		if err != nil {
 			fmt.Printf("Error reading body: %v", err)
 			http.Error(w, "can't read body", http.StatusBadRequest)

internal/topo/source/httppull_source.go

@@ -18,16 +18,17 @@ import (
 	"crypto/md5"
 	"encoding/hex"
 	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/cert"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
-	"io/ioutil"
-	"net/http"
-	"net/url"
-	"strings"
-	"time"
 )
 
 const DEFAULT_INTERVAL = 10000
@@ -223,7 +224,7 @@ func (hps *HTTPPullSource) initTimerPull(ctx api.StreamContext, consumer chan<-
 					logger.Warnf("Found error http return code: %d when trying to reach %v ", resp.StatusCode, hps)
 					break
 				}
-				c, err := ioutil.ReadAll(resp.Body)
+				c, err := io.ReadAll(resp.Body)
 				if err != nil {
 					logger.Warnf("Found error %s when trying to reach %v ", err, hps)
 				}

internal/topo/topotest/mocknode/mock_data.go

@@ -16,10 +16,11 @@ package mocknode
 
 import (
 	"encoding/base64"
+	"os"
+	"path"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/xsql"
-	"io/ioutil"
-	"path"
 )
 
 // TestData The time diff must larger than timeleap
@@ -1010,7 +1011,7 @@ func getImg() ([]byte, string) {
 	if err != nil {
 		conf.Log.Fatalf("Cannot find docs folder: %v", err)
 	}
-	image, err := ioutil.ReadFile(path.Join(docsFolder, "cover.jpg"))
+	image, err := os.ReadFile(path.Join(docsFolder, "cover.jpg"))
 	if err != nil {
 		conf.Log.Fatalf("Cannot read image: %v", err)
 	}

test/plugins/service/server.go

@@ -18,7 +18,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"log"
 	"math/rand"
 	"net/http"
@@ -26,14 +26,14 @@ import (
 )
 
 func alert(w http.ResponseWriter, req *http.Request) {
-	buf, bodyErr := ioutil.ReadAll(req.Body)
+	buf, bodyErr := io.ReadAll(req.Body)
 	if bodyErr != nil {
 		log.Print("bodyErr ", bodyErr.Error())
 		http.Error(w, bodyErr.Error(), http.StatusInternalServerError)
 		return
 	}
 
-	rdr1 := ioutil.NopCloser(bytes.NewBuffer(buf))
+	rdr1 := io.NopCloser(bytes.NewBuffer(buf))
 	log.Printf("BODY: %q", rdr1)
 }
 
@@ -47,7 +47,7 @@ type Sensor struct {
 var s = &Sensor{}
 
 func pullSrv(w http.ResponseWriter, req *http.Request) {
-	buf, bodyErr := ioutil.ReadAll(req.Body)
+	buf, bodyErr := io.ReadAll(req.Body)
 	if bodyErr != nil {
 		log.Print("bodyErr ", bodyErr.Error())
 		http.Error(w, bodyErr.Error(), http.StatusInternalServerError)

tools/kubernetes/conf/conf.go

@@ -18,9 +18,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"gopkg.in/yaml.v3"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"os"
 	"path"
@@ -29,6 +27,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"gopkg.in/yaml.v3"
 )
 
 type (
@@ -82,7 +81,7 @@ func (c *config) initConfig() bool {
 		fmt.Println("conf path err : ", err)
 		return false
 	}
-	sliByte, err := ioutil.ReadFile(confPath)
+	sliByte, err := os.ReadFile(confPath)
 	if nil != err {
 		fmt.Println("load conf err : ", err)
 		return false
@@ -171,7 +170,7 @@ func fetchContents(request *http.Request) (data []byte, err error) {
 		return nil, err
 	}
 	defer respon.Body.Close()
-	data, err = ioutil.ReadAll(respon.Body)
+	data, err = io.ReadAll(respon.Body)
 	if nil != err {
 		return nil, err
 	}
@@ -218,7 +217,7 @@ func Delete(inUrl string) (data []byte, err error) {
 }
 
 func LoadFileUnmarshal(path string, ret interface{}) error {
-	sliByte, err := ioutil.ReadFile(path)
+	sliByte, err := os.ReadFile(path)
 	if nil != err {
 		return err
 	}
@@ -234,5 +233,5 @@ func SaveFileMarshal(path string, content interface{}) error {
 	if nil != err {
 		return err
 	}
-	return ioutil.WriteFile(path, data, 0666)
+	return os.WriteFile(path, data, 0666)
 }

tools/kubernetes/util/util.go

@@ -17,12 +17,12 @@ package util
 import (
 	"encoding/json"
 	"fmt"
-	kconf "github.com/lf-edge/ekuiper/tools/kubernetes/conf"
-	"io/ioutil"
 	"os"
 	"path"
 	"strings"
 	"time"
+
+	kconf "github.com/lf-edge/ekuiper/tools/kubernetes/conf"
 )
 
 type (
@@ -143,12 +143,17 @@ func (s *server) saveHistoryFile() bool {
 	return true
 }
 
-func (s *server) isUpdate(info os.FileInfo) bool {
-	v := s.mapHistoryFile[info.Name()]
+func (s *server) isUpdate(entry os.DirEntry) bool {
+	v := s.mapHistoryFile[entry.Name()]
 	if nil == v {
 		return true
 	}
 
+	info, err := entry.Info()
+	if err != nil {
+		return false
+	}
+
 	if v.LoadTime < info.ModTime().Unix() {
 		return true
 	}
@@ -156,27 +161,27 @@ func (s *server) isUpdate(info os.FileInfo) bool {
 }
 
 func (s *server) processDir() bool {
-	infos, err := ioutil.ReadDir(s.dirCommand)
+	dirEntries, err := os.ReadDir(s.dirCommand)
 	if nil != err {
 		s.logs = append(s.logs, fmt.Sprintf("read command dir:%v", err))
 		return false
 	}
 	conf := kconf.GetConf()
 	host := fmt.Sprintf(`http://%s:%d`, conf.GetIp(), conf.GetPort())
-	for _, info := range infos {
-		if !strings.HasSuffix(info.Name(), ".json") {
+	for _, entry := range dirEntries {
+		if !strings.HasSuffix(entry.Name(), ".json") {
 			continue
 		}
-		if !s.isUpdate(info) {
+		if !s.isUpdate(entry) {
 			continue
 		}
 
 		hisFile := new(historyFile)
-		hisFile.setName(info.Name())
+		hisFile.setName(entry.Name())
 		hisFile.setLoadTime(time.Now().Unix())
-		s.mapHistoryFile[info.Name()] = hisFile
+		s.mapHistoryFile[entry.Name()] = hisFile
 
-		filePath := path.Join(s.dirCommand, info.Name())
+		filePath := path.Join(s.dirCommand, entry.Name())
 		file := new(fileData)
 		err = kconf.LoadFileUnmarshal(filePath, file)
 		if nil != err {