example influxdb event submission using http & udp

influx_event_example.go

package main

import (
	"log"
	"net/http"
	"time"

	ifxClient "github.com/influxdata/influxdb1-client/v2"
)

var influxClient ifxClient.Client

const (
	dbName          = "testing"
	measurementName = "user_register_tb"
)

func registerUserByAdmin1(w http.ResponseWriter, r *http.Request) {
	log.Println("user is registered by admin 1...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_1"

	fields := make(map[string]interface{})
	fields["user"] = "user"

	go sendInfluxEvent(fields, tags)
}

func registerVerificatorByAdmin1(w http.ResponseWriter, r *http.Request) {
	log.Println("admin is registered by admin 1...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_1"

	fields := make(map[string]interface{})
	fields["user"] = "verificator"

	go sendInfluxEvent(fields, tags)
}

func registerUserByAdmin2(w http.ResponseWriter, r *http.Request) {
	log.Println("user is registered by admin 2...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_2"

	fields := make(map[string]interface{})
	fields["user"] = "user"

	go sendInfluxEvent(fields, tags)
}

func registerVerificatorByAdmin2(w http.ResponseWriter, r *http.Request) {
	log.Println("verificator is registered by admin 2...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_2"

	fields := make(map[string]interface{})
	fields["user"] = "verificator"

	go sendInfluxEvent(fields, tags)
}

func sendInfluxEvent(fields map[string]interface{}, tags map[string]string) {
	// Create a new point batch
	bp, err := ifxClient.NewBatchPoints(ifxClient.BatchPointsConfig{
		Database: dbName,
	})
	if err != nil {
		log.Fatal(err.Error())
	}

	pt, err := ifxClient.NewPoint(measurementName, tags, fields, time.Now())
	if err != nil {
		log.Fatal(err.Error())
	}
	bp.AddPoint(pt)

	// Write the batch
	err = influxClient.Write(bp)
	if err != nil {
		log.Fatal(err.Error())
	}
}

func main() {
	var err error

	influxClient, err = ifxClient.NewHTTPClient(ifxClient.HTTPConfig{
		Addr: "http://localhost:8086",
	})
	if err != nil {
		log.Fatalln(err.Error())
	}
	defer influxClient.Close()

	s := http.NewServeMux()
	s.HandleFunc("/admin1/user", registerUserByAdmin1)
	s.HandleFunc("/admin1/verificator", registerVerificatorByAdmin1)
	s.HandleFunc("/admin2/user", registerUserByAdmin2)
	s.HandleFunc("/admin2/verificator", registerVerificatorByAdmin2)

	log.Println("application is running on port 8080...")
	http.ListenAndServe(":8080", s)
}

influx_udp_event_example.go

package main

import (
	"log"
	"net/http"
	"time"

	ifxClient "github.com/influxdata/influxdb1-client/v2"
)

var influxClient ifxClient.Client

const (
	dbName          = "testing"
	measurementName = "user_register_tb"
)

func registerUserByAdmin1(w http.ResponseWriter, r *http.Request) {
	log.Println("user is registered by admin 1...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_1"

	fields := make(map[string]interface{})
	fields["user"] = "user"

	go sendInfluxEvent(fields, tags)
}

func registerVerificatorByAdmin1(w http.ResponseWriter, r *http.Request) {
	log.Println("admin is registered by admin 1...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_1"

	fields := make(map[string]interface{})
	fields["user"] = "verificator"

	go sendInfluxEvent(fields, tags)
}

func registerUserByAdmin2(w http.ResponseWriter, r *http.Request) {
	log.Println("user is registered by admin 2...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_2"

	fields := make(map[string]interface{})
	fields["user"] = "user"

	go sendInfluxEvent(fields, tags)
}

func registerVerificatorByAdmin2(w http.ResponseWriter, r *http.Request) {
	log.Println("verificator is registered by admin 2...")

	tags := make(map[string]string)
	tags["inserted_by"] = "admin_2"

	fields := make(map[string]interface{})
	fields["user"] = "verificator"

	go sendInfluxEvent(fields, tags)
}

func sendInfluxEvent(fields map[string]interface{}, tags map[string]string) {
	// Create a new point batch
	bp, err := ifxClient.NewBatchPoints(ifxClient.BatchPointsConfig{})
	if err != nil {
		log.Fatal(err.Error())
	}

	// udp will use database defined in influx configuration file.
	// udp connection cannot use user defined database in code.
	pt, err := ifxClient.NewPoint(measurementName, tags, fields, time.Now())
	if err != nil {
		log.Fatal(err.Error())
	}
	bp.AddPoint(pt)

	// Write the batch
	err = influxClient.Write(bp)
	if err != nil {
		log.Fatal(err.Error())
	}
}

func main() {
	var err error

	influxClient, err = ifxClient.NewUDPClient(ifxClient.UDPConfig{
		Addr: "localhost:8089", // influxdb default udp port
	})
	if err != nil {
		log.Fatalln(err.Error())
	}
	defer influxClient.Close()

	s := http.NewServeMux()
	s.HandleFunc("/admin1/user", registerUserByAdmin1)
	s.HandleFunc("/admin1/verificator", registerVerificatorByAdmin1)
	s.HandleFunc("/admin2/user", registerUserByAdmin2)
	s.HandleFunc("/admin2/verificator", registerVerificatorByAdmin2)

	log.Println("application is running on port 8080...")
	http.ListenAndServe(":8080", s)
}