// Copyright 2019 github.com. All rights reserved. // Use of this source code is governed by github.com. package model import ( "dust-monitor/consts" "dust-monitor/errors" "encoding/json" "fmt" "strings" "time" "github.com/jaryhe/gopkgs/logger" "go.uber.org/zap" iclient "github.com/influxdata/influxdb/client/v2" "github.com/jaryhe/gopkgs/influxdb" ) func WriteDustData(sn string, tags map[string]string, fields map[string]interface{}, t time.Time, cmdType string) error { // 主动获取数据转换 switch cmdType { case "2111": cmdType = "2011" case "2131": cmdType = "2031" case "2161": cmdType = "2061" } dbName := "dust_monitor_" + cmdType fmt.Printf("xxxxxxxx:%v,%v\n", dbName,sn) bp, err := iclient.NewBatchPoints(iclient.BatchPointsConfig{ Database: dbName, Precision: "ns", }) if err != nil { logger.Error("func", zap.String("call", "NewBatchPoints"), zap.String("args", dbName), zap.String("error", err.Error())) return errors.DataBaseError } pt, err := iclient.NewPoint( sn, tags, fields, t, ) if err != nil { logger.Error("func", zap.String("call", "NewPoint"), zap.String("args", sn), zap.String("error", err.Error())) return errors.DataBaseError } bp.AddPoint(pt) err = influxdb.InfluxCli.Write(bp) if err != nil { logger.Error("func", zap.String("call", "InfluxCli.Write"), zap.String("args", sn), zap.String("error", err.Error())) if strings.Contains(err.Error(), consts.DatabaseNotFound) { Create(influxdb.InfluxCli, dbName) err = influxdb.InfluxCli.Write(bp) if err != nil { return errors.DataBaseError } return nil } return errors.DataBaseError } return nil } func Query(sql string, db string, result interface{}) ([]map[string]interface{}, error) { qt := iclient.Query{ Database: db, Command: sql, } r, err := influxdb.InfluxCli.Query(qt) if err != nil { logger.Error("func", zap.String("call", "Query "+ db), zap.String("args", sql), zap.String("error", err.Error())) return nil, err } if r == nil { return nil, nil } if len(r.Results) == 0 { return nil, nil } if len(r.Results[0].Series) == 0 { return nil, nil } colNames := r.Results[0].Series[0].Columns var marray = make([]map[string]interface{}, len(r.Results[0].Series[0].Values)) for i, row := range r.Results[0].Series[0].Values { item := map[string]interface{}{} for j, v := range row { if colNames[j] == "time" { t, _ := time.Parse(time.RFC3339, v.(string)) v = t.Format("2006-01-02 15:04:05") } item[colNames[j]] = v } marray[i] = item } if result != nil { bytes, _ := json.Marshal(marray) err = json.Unmarshal(bytes, result) if err != nil { return nil, err } } return marray, nil } func Create(conn iclient.Client, db string) error { qt := iclient.Query{ Command: "create database " + db, } _, err := conn.Query(qt) if err != nil { return err } return nil } func WriteDustDatas(sn string, tags map[string]string, fields []map[string]interface{}, t []time.Time, cmdType string) error { // 主动获取数据转换 switch cmdType { case "2111": cmdType = "2011" case "2131": cmdType = "2031" case "2161": cmdType = "2061" } if len(fields) != len(t) { return errors.ParamsError } dbName := "dust_monitor_" + cmdType bp, err := iclient.NewBatchPoints(iclient.BatchPointsConfig{ Database: dbName, Precision: "ns", }) if err != nil { logger.Error("func", zap.String("call", "NewBatchPoints"), zap.String("args", dbName), zap.String("error", err.Error())) return errors.DataBaseError } pts := make([]*iclient.Point, len(fields)) for i, _ := range fields { pt, err := iclient.NewPoint( sn, tags, fields[i], t[i], ) if err != nil { logger.Error("func", zap.String("call", "NewPoint"), zap.String("args", sn), zap.String("error", err.Error())) return errors.DataBaseError } pts[i] = pt } bp.AddPoints(pts) err = influxdb.InfluxCli.Write(bp) if err != nil { logger.Error("func", zap.String("call", "InfluxCli.Write"), zap.String("args", sn), zap.String("error", err.Error())) if strings.Contains(err.Error(), consts.DatabaseNotFound) { Create(influxdb.InfluxCli, dbName) err = influxdb.InfluxCli.Write(bp) if err != nil { return errors.DataBaseError } return nil } return errors.DataBaseError } return nil }