tower-monitor/model/tower.go

// Copyright 2019 github.com. All rights reserved.
// Use of this source code is governed by github.com.

package model

import (
	"tower-monitor/consts"
	"tower-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 WriteTowerData(sn string, tags map[string]string, fields map[string]interface{}, t time.Time, cmdType string) error {
	dbName := "tower_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
}