feat: implement OPC UA mapper using v1beta1 mapper-framework

mappers/Kubeedge-v1.22.0/opcua-mapper/Dockerfile_nostream

@@ -0,0 +1,21 @@
+FROM golang:1.23-alpine3.19 AS builder
+
+WORKDIR /build
+
+ENV GO111MODULE=on \
+    GOPROXY=https://goproxy.cn,direct
+
+COPY . .
+
+RUN CGO_ENABLED=0 GOOS=linux go build -o main cmd/main.go
+
+
+FROM ubuntu:18.04
+
+RUN mkdir -p kubeedge
+
+COPY --from=builder /build/main kubeedge/
+COPY ./config.yaml kubeedge/
+
+WORKDIR kubeedge
+

mappers/Kubeedge-v1.22.0/opcua-mapper/Makefile

@@ -0,0 +1,34 @@
+SHELL := /bin/bash
+
+curr_dir := $(patsubst %/,%,$(dir $(abspath $(lastword $(MAKEFILE_LIST)))))
+rest_args := $(wordlist 2, $(words $(MAKECMDGOALS)), $(MAKECMDGOALS))
+$(eval $(rest_args):;@:)
+
+help:
+	#
+	# Usage:
+	#   make generate :  generate a mapper based on a template.
+	#   make mapper {mapper-name} <action> <parameter>:  execute mapper building process.
+	#
+	# Actions:
+	#   -           mod, m  :  download code dependencies.
+	#   -          lint, l  :  verify code via go fmt and `golangci-lint`.
+	#   -         build, b  :  compile code.
+	#   -       package, p  :  package docker image.
+	#   -         clean, c  :  clean output binary.
+	#
+	# Parameters:
+	#   ARM   : true or undefined
+	#   ARM64 : true or undefined
+	#
+	# Example:
+	#   -  make mapper modbus ARM64=true :  execute `build` "modbus" mapper for ARM64.
+	#   -        make mapper modbus test :  execute `test` "modbus" mapper.
+	@echo
+
+make_rules := $(shell ls $(curr_dir)/hack/make-rules | sed 's/.sh//g')
+$(make_rules):
+	@$(curr_dir)/hack/make-rules/$@.sh $(rest_args)
+
+.DEFAULT_GOAL := help
+.PHONY: $(make_rules) build test package

mappers/Kubeedge-v1.22.0/opcua-mapper/cmd/main.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"errors"
+
+	"k8s.io/klog/v2"
+
+	"github.com/kubeedge/mapper-framework/pkg/common"
+	"github.com/kubeedge/mapper-framework/pkg/config"
+	"github.com/kubeedge/mapper-framework/pkg/grpcclient"
+	"github.com/kubeedge/mapper-framework/pkg/grpcserver"
+	"github.com/kubeedge/mapper-framework/pkg/httpserver"
+	"github.com/kubeedge/mappers-go/mappers/kubeedge-v1.22.0/opcua-mapper/device"
+)
+
+func main() {
+	var err error
+	var c *config.Config
+
+	klog.InitFlags(nil)
+	defer klog.Flush()
+
+	if c, err = config.Parse(); err != nil {
+		klog.Fatal(err)
+	}
+	klog.Infof("config: %+v", c)
+
+	klog.Infoln("Mapper will register to edgecore")
+	deviceList, deviceModelList, err := grpcclient.RegisterMapper(true)
+	if err != nil {
+		klog.Fatal(err)
+	}
+	klog.Infoln("Mapper register finished")
+
+	panel := device.NewDevPanel()
+	err = panel.DevInit(deviceList, deviceModelList)
+	if err != nil && !errors.Is(err, device.ErrEmptyData) {
+		klog.Fatal(err)
+	}
+	klog.Infoln("devInit finished")
+	go panel.DevStart()
+
+	// start http server
+	httpServer := httpserver.NewRestServer(panel, c.Common.HTTPPort)
+	go httpServer.StartServer()
+
+	// start grpc server
+	grpcServer := grpcserver.NewServer(
+		grpcserver.Config{
+			SockPath: c.GrpcServer.SocketPath,
+			Protocol: common.ProtocolCustomized,
+		},
+		panel,
+	)
+	defer grpcServer.Stop()
+	if err = grpcServer.Start(); err != nil {
+		klog.Fatal(err)
+	}
+
+}

mappers/Kubeedge-v1.22.0/opcua-mapper/config.yaml

@@ -0,0 +1,9 @@
+grpc_server:
+  socket_path: /etc/kubeedge/opcua-mapper.sock
+common:
+  name: opcua-mapper
+  version: v1.13.0
+  api_version: v1beta1
+  protocol:  opcua # TODO add your protocol name
+  address: 127.0.0.1:1234 # TODO add your protocol address
+  edgecore_sock: /etc/kubeedge/dmi.sock

mappers/Kubeedge-v1.22.0/opcua-mapper/data/dbmethod/influxdb2/client.go

@@ -0,0 +1,76 @@
+package influxdb2
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+	"github.com/kubeedge/mapper-framework/pkg/common"
+)
+
+type DataBaseConfig struct {
+	Influxdb2ClientConfig *Influxdb2ClientConfig `json:"influxdb2ClientConfig,omitempty"`
+	Influxdb2DataConfig   *Influxdb2DataConfig   `json:"influxdb2DataConfig,omitempty"`
+}
+
+type Influxdb2ClientConfig struct {
+	Url    string `json:"url,omitempty"`
+	Org    string `json:"org,omitempty"`
+	Bucket string `json:"bucket,omitempty"`
+}
+
+type Influxdb2DataConfig struct {
+	Measurement string            `json:"measurement,omitempty"`
+	Tag         map[string]string `json:"tag,omitempty"`
+	FieldKey    string            `json:"fieldKey,omitempty"`
+}
+
+func NewDataBaseClient(clientConfig json.RawMessage, dataConfig json.RawMessage) (*DataBaseConfig, error) {
+	// parse influx database config data
+	influxdb2ClientConfig := new(Influxdb2ClientConfig)
+	influxdb2DataConfig := new(Influxdb2DataConfig)
+	err := json.Unmarshal(clientConfig, influxdb2ClientConfig)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(dataConfig, influxdb2DataConfig)
+	if err != nil {
+		return nil, err
+	}
+	return &DataBaseConfig{
+		Influxdb2ClientConfig: influxdb2ClientConfig,
+		Influxdb2DataConfig:   influxdb2DataConfig,
+	}, nil
+}
+
+func (d *DataBaseConfig) InitDbClient() influxdb2.Client {
+	var usrtoken string
+	usrtoken = os.Getenv("TOKEN")
+	client := influxdb2.NewClient(d.Influxdb2ClientConfig.Url, usrtoken)
+
+	return client
+}
+
+func (d *DataBaseConfig) CloseSession(client influxdb2.Client) {
+	client.Close()
+}
+
+func (d *DataBaseConfig) AddData(data *common.DataModel, client influxdb2.Client) error {
+	// write device data to influx database
+	writeAPI := client.WriteAPIBlocking(d.Influxdb2ClientConfig.Org, d.Influxdb2ClientConfig.Bucket)
+	p := influxdb2.NewPoint(d.Influxdb2DataConfig.Measurement,
+		d.Influxdb2DataConfig.Tag,
+		map[string]interface{}{d.Influxdb2DataConfig.FieldKey: data.Value},
+		time.Now())
+	// write point immediately
+	err := writeAPI.WritePoint(context.Background(), p)
+	if err != nil {
+		klog.V(4).Info("Exit AddData")
+		return err
+	}
+	return nil
+}

mappers/Kubeedge-v1.22.0/opcua-mapper/data/dbmethod/influxdb2/handler.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2023 The KubeEdge Authors.
+
+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 influxdb2
+
+import (
+	"context"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	"github.com/kubeedge/mappers-go/mappers/kubeedge-v1.22.0/opcua-mapper/driver"
+	"github.com/kubeedge/mapper-framework/pkg/common"
+)
+
+func DataHandler(ctx context.Context, twin *common.Twin, client *driver.CustomizedClient, visitorConfig *driver.VisitorConfig, dataModel *common.DataModel) {
+	dbConfig, err := NewDataBaseClient(twin.Property.PushMethod.DBMethod.DBConfig.Influxdb2ClientConfig, twin.Property.PushMethod.DBMethod.DBConfig.Influxdb2DataConfig)
+	if err != nil {
+		klog.Errorf("new database client error: %v", err)
+		return
+	}
+	dbClient := dbConfig.InitDbClient()
+	if err != nil {
+		klog.Errorf("init database client err: %v", err)
+		return
+	}
+	reportCycle := time.Millisecond * time.Duration(twin.Property.ReportCycle)
+	if reportCycle == 0 {
+		reportCycle = common.DefaultReportCycle
+	}
+	ticker := time.NewTicker(reportCycle)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				deviceData, err := client.GetDeviceData(visitorConfig)
+				if err != nil {
+					klog.Errorf("publish error: %v", err)
+					continue
+				}
+				sData, err := common.ConvertToString(deviceData)
+				if err != nil {
+					klog.Errorf("Failed to convert publish method data : %v", err)
+					continue
+				}
+				dataModel.SetValue(sData)
+				dataModel.SetTimeStamp()
+
+				err = dbConfig.AddData(dataModel, dbClient)
+				if err != nil {
+					klog.Errorf("influx database add data error: %v", err)
+					return
+				}
+			case <-ctx.Done():
+				dbConfig.CloseSession(dbClient)
+				return
+			}
+		}
+	}()
+}

mappers/Kubeedge-v1.22.0/opcua-mapper/data/dbmethod/mysql/client.go

@@ -0,0 +1,105 @@
+/*
+Copyright 2024 The KubeEdge Authors.
+
+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 mysql
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"os"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+	"k8s.io/klog/v2"
+
+	"github.com/kubeedge/mapper-framework/pkg/common"
+)
+
+var (
+	DB *sql.DB
+)
+
+type DataBaseConfig struct {
+	MySQLClientConfig *MySQLClientConfig `json:"mysqlClientConfig"`
+}
+
+type MySQLClientConfig struct {
+	Addr     string `json:"addr,omitempty"`
+	Database string `json:"database,omitempty"`
+	UserName string `json:"userName,omitempty"`
+}
+
+func NewDataBaseClient(config json.RawMessage) (*DataBaseConfig, error) {
+	configdata := new(MySQLClientConfig)
+	err := json.Unmarshal(config, configdata)
+	if err != nil {
+		return nil, err
+	}
+	return &DataBaseConfig{
+		MySQLClientConfig: configdata,
+	}, nil
+}
+
+func (d *DataBaseConfig) InitDbClient() error {
+	password := os.Getenv("PASSWORD")
+	usrName := d.MySQLClientConfig.UserName
+	addr := d.MySQLClientConfig.Addr
+	dataBase := d.MySQLClientConfig.Database
+	dataSourceName := fmt.Sprintf("%s:%s@tcp(%s)/%s", usrName, password, addr, dataBase)
+	var err error
+	DB, err = sql.Open("mysql", dataSourceName)
+	if err != nil {
+		return fmt.Errorf("connection to %s of mysql faild with err:%v", dataBase, err)
+	}
+
+	return nil
+}
+
+func (d *DataBaseConfig) CloseSession() {
+	err := DB.Close()
+	if err != nil {
+		klog.Errorf("close mysql failed with err:%v", err)
+	}
+}
+
+func (d *DataBaseConfig) AddData(data *common.DataModel) error {
+	tableName := data.Namespace + "/" + data.DeviceName + "/" + data.PropertyName
+	datatime := time.Unix(data.TimeStamp/1e3, 0).Format("2006-01-02 15:04:05")
+
+	createTable := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (id INT AUTO_INCREMENT PRIMARY KEY, ts  DATETIME NOT NULL,field TEXT)", tableName)
+	_, err := DB.Exec(createTable)
+	if err != nil {
+		return fmt.Errorf("create tabe into mysql failed with err:%v", err)
+	}
+
+	stmt, err := DB.Prepare(fmt.Sprintf("INSERT INTO `%s` (ts,field) VALUES (?,?)", tableName))
+	if err != nil {
+		return fmt.Errorf("prepare parament failed with err:%v", err)
+	}
+	defer func(stmt *sql.Stmt) {
+		err := stmt.Close()
+		if err != nil {
+			klog.Errorf("close mysql's statement failed with err:%v", err)
+		}
+	}(stmt)
+	_, err = stmt.Exec(datatime, data.Value)
+	if err != nil {
+		return fmt.Errorf("insert data into msyql failed with err:%v", err)
+	}
+
+	return nil
+}