Merge pull request #130 from fuchendou/main

Provide a Common MQTT Mapper

.gitignore

@@ -3,3 +3,12 @@ pkg/modbus/modbus
 pkg/bluetooth/bluetooth
 .idea
 vendor/
+.lightly
+.c_bin
+.classes
+logs
+*.log
+node_modules
+dist
+dist-ssr
+*.local

mappers/device-v1beta1/mqtt-mapper/Dockerfile_nostream

@@ -0,0 +1,20 @@
+FROM golang:1.21.11-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/device-v1beta1/mqtt-mapper/Dockerfile_stream

@@ -0,0 +1,35 @@
+FROM golang:1.21.11-bullseye AS builder
+
+WORKDIR /build
+
+ENV GO111MODULE=on \
+    GOPROXY=https://goproxy.cn,direct
+
+COPY . .
+
+RUN apt-get update && \
+    apt-get install -y bzip2 curl upx-ucl gcc-aarch64-linux-gnu libc6-dev-arm64-cross gcc-arm-linux-gnueabi libc6-dev-armel-cross libva-dev libva-drm2 libx11-dev libvdpau-dev libxext-dev libsdl1.2-dev libxcb1-dev libxau-dev libxdmcp-dev yasm
+
+RUN curl -sLO https://ffmpeg.org/releases/ffmpeg-4.1.6.tar.bz2 && \
+    tar -jx --strip-components=1 -f ffmpeg-4.1.6.tar.bz2 &&  \
+    ./configure &&  make && \
+    make install
+
+RUN GOOS=linux go build -o main cmd/main.go
+
+FROM ubuntu:18.04
+
+RUN mkdir -p kubeedge
+
+RUN apt-get update && \
+    apt-get install -y bzip2 curl upx-ucl gcc-aarch64-linux-gnu libc6-dev-arm64-cross gcc-arm-linux-gnueabi libc6-dev-armel-cross libva-dev libva-drm2 libx11-dev libvdpau-dev libxext-dev libsdl1.2-dev libxcb1-dev libxau-dev libxdmcp-dev yasm
+
+RUN curl -sLO https://ffmpeg.org/releases/ffmpeg-4.1.6.tar.bz2 && \
+    tar -jx --strip-components=1 -f ffmpeg-4.1.6.tar.bz2 &&  \
+    ./configure &&  make && \
+    make install
+
+COPY --from=builder /build/main kubeedge/
+COPY ./config.yaml kubeedge/
+
+WORKDIR kubeedge

mappers/device-v1beta1/mqtt-mapper/cmd/main.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"errors"
+
+	"k8s.io/klog/v2"
+
+	"github.com/kubeedge/mqtt/device"
+	"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"
+)
+
+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/device-v1beta1/mqtt-mapper/config.yaml

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

mappers/device-v1beta1/mqtt-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/device-v1beta1/mqtt-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/mqtt/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
+			}
+		}
+	}()
+}