-
Notifications
You must be signed in to change notification settings - Fork 67
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Provide a Common MQTT Mapper #130
Open
fuchendou
wants to merge
5
commits into
kubeedge:main
Choose a base branch
from
fuchendou:main
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
5 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,20 @@ | ||
FROM golang:1.20.10-alpine3.18 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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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/[email protected] $(rest_args) | ||
|
||
.DEFAULT_GOAL := help | ||
.PHONY: $(make_rules) build test package |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,60 @@ | ||
package main | ||
|
||
import ( | ||
"errors" | ||
|
||
"k8s.io/klog/v2" | ||
|
||
"github.com/kubeedge/kubeedge-v1.17.0/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) | ||
} | ||
|
||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,9 @@ | ||
grpc_server: | ||
socket_path: /etc/kubeedge/kubeedge-v1.17.0.sock | ||
common: | ||
name: github.com/kubeedge/mqtt-mapper | ||
version: v1.13.0 | ||
api_version: v1.0.0 | ||
protocol: mqtt | ||
address: 127.0.0.1 | ||
edgecore_sock: /etc/kubeedge/dmi.sock |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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/kubeedge-v1.17.0/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 | ||
} | ||
} | ||
}() | ||
} |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
remove this file