init

LICENSE

@@ -0,0 +1,29 @@
+BSD 3-Clause License
+
+Copyright (c) 2016, vadv
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+* Neither the name of the copyright holder nor the names of its
+  contributors may be used to endorse or promote products derived from
+  this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Makefile

@@ -0,0 +1,12 @@
+BINARY=./bin/zabbix-bench
+SOURCEDIR=./src
+SOURCES := $(shell find $(SOURCEDIR) -name '*.go')
+GOPATH := ${PWD}:${GOPATH}
+export GOPATH
+.DEFAULT_GOAL: $(BINARY)
+$(BINARY): $(SOURCES)
+	go build -o ${BINARY} $(SOURCEDIR)/main.go
+run: clean $(BINARY)
+	${BINARY}
+clean:
+	rm -f $(BINARY)

README.md

@@ -0,0 +1,19 @@
+Benchmarking and stress testing tool for the Zabbix server
+
+```bash
+Usage of zabbix-bench:
+  -client int
+        number of concurrent clients (default 200)
+  -client-format string
+        format of client name (default "client-%d")
+  -metric-format string
+        format of metric name in packet (default "metric-%d")
+  -packet-delay duration
+        delay of send packet (default 100ms)
+  -packet-send-timeout duration
+        packet send timeout (default 10ms)
+  -packet-size int
+        count of metric in packet (default 100)
+  -zabbix string
+        address of zabbix server (default "127.0.0.1:10051")
+```

src/main.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"runtime"
+	"sync"
+	"syscall"
+	"time"
+	"zabbix"
+)
+
+var (
+	argClientCount = flag.Int("client", 200, "number of concurrent clients")
+	argClinetName  = flag.String("client-format", "client-%d", "format of client name")
+	argPacketSize  = flag.Int("packet-size", 100, "count of metric in packet")
+	argMetricName  = flag.String("metric-format", "metric-%d", "format of metric name in packet")
+	argPacketDelay = flag.Duration("packet-delay", 100*time.Millisecond, "delay of send packet")
+	argSendTimeout = flag.Duration("packet-send-timeout", 10*time.Millisecond, "packet send timeout")
+	argZabbix      = flag.String("zabbix", "127.0.0.1:10051", "address of zabbix server")
+
+	errorChannel     = make(chan error, 10)
+	completedChannel = make(chan int, 10)
+	signalChannel    = make(chan os.Signal, 1)
+
+	mutex               = &sync.Mutex{}
+	counter, total, sec = 0, 0, 1
+)
+
+func main() {
+
+	runtime.GOMAXPROCS(runtime.NumCPU() * 4)
+	signal.Notify(signalChannel, os.Interrupt)
+	signal.Notify(signalChannel, syscall.SIGTERM)
+
+	if !flag.Parsed() {
+		flag.Parse()
+	}
+
+	for i := 0; i < *argClientCount; i++ {
+		go StartClient(i)
+	}
+
+	ticker := time.Tick(time.Second)
+	for {
+		select {
+		case <-ticker:
+			mutex.Lock()
+			sec += 1
+			os.Stdout.WriteString(fmt.Sprintf("Metric sended: %d\n", counter))
+			counter = 0
+			mutex.Unlock()
+		case count := <-completedChannel:
+			mutex.Lock()
+			counter += count
+			total += count
+			mutex.Unlock()
+		case err := <-errorChannel:
+			os.Stderr.WriteString(fmt.Sprintf("Error write package:\t%s\n", err.Error()))
+		case <-signalChannel:
+			os.Stdout.WriteString(fmt.Sprintf("Total: %d (%d metric/s)\n", total, int(total/sec)))
+			os.Exit(0)
+		}
+	}
+
+}
+
+// client of zabbix server
+type client struct {
+	id     int
+	host   string
+	sender *zabbix.Sender
+}
+
+// generate and send zabbix packet
+func (c *client) send() error {
+	now := time.Now().Unix()
+	metrics := make([]*zabbix.Metric, 0)
+	for i := 0; i < *argPacketSize; i++ {
+		metrics = append(metrics, zabbix.NewMetric(c.host, fmt.Sprintf(*argMetricName, i), string(i), now))
+	}
+	return c.sender.Send(zabbix.NewPacket(metrics, now))
+}
+
+func StartClient(id int) {
+	c := &client{
+		id:     id,
+		host:   fmt.Sprintf(*argClinetName, id),
+		sender: zabbix.NewSender(*argZabbix),
+	}
+	ticker := time.Tick(*argPacketDelay)
+	for {
+		select {
+		case <-ticker:
+			if err := c.send(); err != nil {
+				errorChannel <- err
+			} else {
+				completedChannel <- *argPacketSize
+			}
+		}
+	}
+}

src/zabbix/metric.go

@@ -0,0 +1,22 @@
+package zabbix
+
+import (
+	"time"
+)
+
+type Metric struct {
+	Host  string `json:"host"`
+	Key   string `json:"key"`
+	Value string `json:"value"`
+	Clock int64  `json:"clock"`
+}
+
+func NewMetric(host, key, value string, clock ...int64) *Metric {
+	m := &Metric{Host: host, Key: key, Value: value}
+	if len(clock) > 0 {
+		m.Clock = clock[0]
+	} else {
+		m.Clock = time.Now().Unix()
+	}
+	return m
+}

src/zabbix/packet.go

@@ -0,0 +1,46 @@
+package zabbix
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	"time"
+)
+
+type Packet struct {
+	Request  string    `json:"request"`
+	Data     []*Metric `json:"data"`
+	Clock    int64     `json:"clock"`
+	jsonData []byte    // cached json data
+	dataLen  []byte    // cached length data
+}
+
+func NewPacket(data []*Metric, clock ...int64) *Packet {
+	p := &Packet{Request: `sender data`, Data: data}
+	if len(clock) > 0 {
+		p.Clock = clock[0]
+	} else {
+		p.Clock = time.Now().Unix()
+	}
+	return p
+}
+
+// cache json data
+func (p *Packet) Json() []byte {
+	if len(p.jsonData) != 0 {
+		return p.jsonData
+	}
+	jsonData, _ := json.Marshal(p)
+	p.jsonData = jsonData
+	return p.jsonData
+}
+
+// cached length data
+func (p *Packet) DataLen() []byte {
+	if len(p.dataLen) > 0 {
+		return p.dataLen
+	}
+	dataLen := make([]byte, 8)
+	binary.LittleEndian.PutUint32(dataLen, uint32(len(p.Json())))
+	p.dataLen = dataLen
+	return p.dataLen
+}

src/zabbix/sender.go

@@ -0,0 +1,69 @@
+package zabbix
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"os"
+)
+
+var zbxHeader = []byte("ZBXD\x01")
+
+type Sender struct {
+	address string
+	iaddr   *net.TCPAddr
+}
+
+func NewSender(address string) *Sender {
+	s := &Sender{address: address}
+	if err := s.resolv(); err != nil {
+		os.Stderr.WriteString(fmt.Sprintf("Can't resolv zabbix address: %s\n", err.Error()))
+		os.Exit(1)
+	}
+	return s
+}
+
+func (s *Sender) resolv() error {
+	if iaddr, err := net.ResolveTCPAddr("tcp", s.address); err != nil {
+		return err
+	} else {
+		s.iaddr = iaddr
+	}
+	return nil
+}
+
+func (s *Sender) connect() (*net.TCPConn, error) {
+	if conn, err := net.DialTCP("tcp", nil, s.iaddr); err != nil { // TODO: timeout
+		return nil, err
+	} else {
+		return conn, nil
+	}
+}
+
+func (s *Sender) read(conn *net.TCPConn) ([]byte, error) {
+	return ioutil.ReadAll(conn) // TODO: timeout
+}
+
+func (s *Sender) Send(packet *Packet) error {
+
+	conn, err := s.connect()
+	if err != nil {
+		return err
+	}
+
+	buffer := append(zbxHeader, packet.DataLen()...)
+	buffer = append(buffer, packet.Json()...)
+
+	_, err = conn.Write(buffer)
+	if err != nil {
+		return err
+	}
+
+	_, err = s.read(conn)
+	if err != nil {
+		os.Stderr.WriteString(fmt.Sprintf("Read zabbix response error: %s\n", err.Error()))
+	}
+
+	return nil
+
+}