Go语言基础（十七）

一、Elastic search

二、Kibana

三、Log Transfer

四、goputils

五、influxDB

六、prometheus

七、granfana

一、Elastic search

1、ES介绍

Elastic search(ES)是一个基于Lucene构建开源的、分布式、RESTful接口的全文搜索引擎，还是一个分布式文档数据库，其中每个字段均可被索引，而且每个字段均可被索引，ES能够向扩展至数以百计的服务器以及处理PB级的数据，可以再极短的时间内存储、搜索、分析大量数据，通常作为具有复杂搜索场景情况下的核心发动机。

ES能做什么？

网上商店，可以使用ES来存储整个产品目录和库存信息，为客户提供精准搜索，可以为客户推荐相关商品；
收集日志或交易数据。需要分析和挖掘数据，寻找趋势，进行统计或总结，或发现异常，可以使用Logstash或者其他工具来进行收集数据，存储到ES中。
github。Github的搜索是基于ES构建的。

2、ES安装

ES下载地址：https://www.elastic.co/cn/downloads/elasticsearch

下载完成之后解压，在cmd中切换到ES的bin目录下，执行启动命令：

elasticsearch.bat，默认会在本机9200端口启动服务。然后就可以在本地浏览器打开：

127.0.0.1:9200

程序员还是比较有情怀的~！

3、ES初识

a、查看健康状况：

curl -X GET 127.0.0.1:9200/_cat/health?v

http://127.0.0.1:9200/_cat/health?v

b、创建索引

curl -X PUT 127.0.0.1:9200/www

c、查询当前ES集群中所有的indices

curl -X GET 127.0.0.1:9200/_cat/indices?v

d、删除索引

curl -X DELETE 127.0.0.1:9200/www

e、插入记录

使用postman

f、检索

ES检索语法比较特别，使用GET方法携带的JSON格式的查询条件。

全检索:

curl -X GET 127.0.0.1/student/go/_search

按条件检索：

curl -H "ContextType:application/json" -X PUT 127.0.0.1:9200/student/go/4 -d '
{
     "query":{
           "match":{"name": "mogul"},
           "size":2
      }
}'

ES默认一次最多返回10条结果，也可以通过设置size字段来设置返回结果的数目。

4、Go连接ES

使用第三方库连接并进行操作，安装步骤如下：

SET GO111MODULE=on
SET GOPROXY=https://goproxy.cn
go get github.com/olivere/elastic/v7

es insert demo:

package main

import (
	"fmt"

	"github.com/olivere/elastic"
	"golang.org/x/net/context"
)

type Student struct {
	Name    string `json:"name"`
	Age     int    `json:"age"`
	Married bool   `json:"married"`
}

func main() {
	// 1初始化连接，得到一个client
	client, err := elastic.NewClient(elastic.SetURL("http://127.0.0.1:9200"))
	if err != nil {
		panic(err)
	}
	fmt.Println("connect to es success")
	p1 := Student{Name: "mogul", Age: 25, Married: false}
	// 链式操作
	put1, err := client.Index().Index("student").Type("go").BodyJson(p1).Do(context.Background())
	if err != nil {
		panic(err)
	}
	fmt.Printf("index student %s to index %s, type %s\n", put1.Id, put1.Index, put1.Type)
}

二、Kibana

Kibana是为 Elasticsearch设计的开源分析和可视化平台。你可以使用 Kibana 来搜索，查看存储在 Elasticsearch 索引中的数据并与之交互。你可以很容易实现高级的数据分析和可视化，以图标的形式展现出来。

1、Kibana安装

Kibana下载地址：https://www.elastic.co/cn/kibana

注意事项：Kibana与Elasticsearch的版本要相互对应，否则可能不兼容。例如

Elasticsearch是7.6.0版本，那么kibana也要是7.6.0。

安装完成之后，修改配置文件，打开kibana中的config中的kibana.yml文件。

2、Kibana运行

切换到kibana的bin目录下，执行kibana.bat，运行的比较慢，耐心等待就可以了，执行完成见下图：

使用本地浏览器打开http://localhost:5601 这个加载的也比较慢，耐心等待~

三、Log Transfer

Log Transfer的目的，就是将日志数据从kafka取出来并发送到ES。

1、加载配置文件

conf.ini

[kafka]
address=192.168.0.108:9200
topic=web_log


[es]
address=127.0.0.1:9200
chansize=100000
nums=16

conf.go

package conf

type LogTransferCfg struct {
	KafkaCfg `ini:"kafka"`
	ESCfg    `ini:"es"`
}

type KafkaCfg struct {
	Address string `ini:"address"`
	Topic   string `ini:"topic"`
}

type ESCfg struct {
	Address string `ini:"address"`
	ChanSize int `ini:"chansize"`
	Nums int `ini:"nums"`
}

2、Kafka

package kafka

import (
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"studygo/day08/logtransfer/es"
)

type LogData struct {
	Data string `json:"data"`
}

func Init(addr []string, topic string) (error) {
	consumer, err := sarama.NewConsumer(addr, nil)
	if err != nil {
		fmt.Printf("fail to start consumer, err:%v\n", err)
		return err
	}
	partitionList, err := consumer.Partitions(topic)  // 根据topic取到所有分区
	if err != nil {
		fmt.Printf("fail to get list of partition:err%v\n", err)
		return err
	}
	fmt.Println("分区列表：", partitionList)
	for partition := range partitionList {
		//针对每个分区创建一个对应的分区消费者
		pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d, err:%v\n", partition, err)
			return err
		}
		defer pc.AsyncClose()
		// 异步从每个分区消费消息
		go func(sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d key:%v value:%v\n", msg.Partition, msg.Offset, msg.Key, msg.Value)

				err := json.Unmarshal(msg.Value, ld)
				if err != nil {
					fmt.Printf("unmarshal failed, err:%v\n", err)
					continue
				}
				ld := es.LogData{Topic: topic, Data:string(msg.Value)}
				es.SendToESChan(&ld)
			}
		}(pc)
	}
	return  err
}

3、ES

package es

import (
	"fmt"
	"github.com/olivere/elastic"
	"golang.org/x/net/context"
	"strings"
	"time"
)

var (
	client *elastic.Client
	ch chan *LogData
)

type LogData struct {
	Topic string `json:"topic"`
	Data string `json:"data"`
}

func Init(address string, chanSize int, nums int) (err error) {
	if !strings.HasPrefix(address, "http://") {
		address="http://" + address
	}
	client, err = elastic.NewClient(elastic.SetURL("http://127.0.0.1:9200"))
	if err != nil {
		return err
	}
	fmt.Println("connect to es success")
	ch = make(chan *LogData, chanSize)
	for i:=; i<nums; i++ {
		go SendToES()
	}
	return err
}

func SendToESChan(msg *LogData) {
	ch <- msg
}

func SendToES() {
	for {
		select {
		case msg := <- ch:
			put1, err := client.Index().Index(msg.Topic).Type("xxx").BodyJson(msg).Do(context.Background())
			if err != nil {
				fmt.Println(err)
			}
			fmt.Printf("index student %v to index %s, type:%s\n", put1.Id, put1.Index, put1.Type)
		default:
			time.Sleep(time.Second)
		}

	}
}

4、LogTransfer实现

package main

import (
	"fmt"
	"studygo/day08/logtransfer/conf"
	"studygo/day08/logtransfer/es"
	"studygo/day08/logtransfer/kafka"

	"gopkg.in/ini.v1"
)

// log transfer
// 将日志数据从kafka取出来发到ES
func main() {
	// 0加载配置文件
	var cfg = new(conf.LogTransferCfg)
	err := ini.MapTo(cfg, "./conf/cfg.ini")
	if err != nil {
		fmt.Printf("init config failed, err:%v\n", err)
		return
	}
	fmt.Printf("cfg:%v\n", cfg)

	// 1初始化es连接的client
	err = es.Init(cfg.ESCfg.Address, cfg.ESCfg.ChanSize, cfg.ESCfg.Nums)
	if err != nil {
		fmt.Printf("init es client failed, err:%v\n", err)
		return
	}
	fmt.Println("init es success.")
	// 2初始化
	// 2.1连接kafka，创建分区消费者
	// 2.2每个分区消费者分别取出数据，通过sendtoes（）将数据发往es
	kafka.Init([]string{cfg.KafkaCfg.Address}, cfg.KafkaCfg.Topic)
	if err != nil {
		fmt.Printf("init kafka consumer failed, err:%v\n", err)
		return
	}

	select {

	}
	// 2从kafka取日志数据
	// 3发往ES
}

四、gopsutil

psutil是一个跨平台进程和系统监视的python库，而gopsutil是go语言版本的实现。

1、goputil安装

go get github.com/shirou/gopsutil

2、采集cpu相关信息

// 采集cpu相关信息
func getCpuInfo() {
	cpuInfo, err := cpu.Info()
	if err != nil {
		fmt.Printf("get cpu info failed, err:%v\n", err)
	}
	for _, c1 := range cpuInfo {
		fmt.Println(c1)
		println()
	}
	// cpu使用率
	for {
		percent, _ := cpu.Percent(time.Second, false)
		fmt.Printf("cpu percent:%v\n", percent)
	}
}

3、cpu负载信息

// 获取cpu负载信息
func getCpuLoad() {
	info, _ := load.Avg()
	fmt.Printf("cpu负载：%v\n", info)
}

4、Memory

func getMemInfo() {
	memInfo, _ := mem.VirtualMemory()
	fmt.Printf("men info:%v\n", memInfo)
}

5、Host

func getHostInfo() {
	hostInfo, _ := host.Info()
	fmt.Printf("host info:%v, uptime:%v, booltime:%v\n", hostInfo, hostInfo.Uptime, hostInfo.BootTime)
}

6、Disk

func getDiskInfo() {
	parts, err := disk.Partitions(true)
	if err != nil {
		fmt.Printf("get partitions failed, err:%v\n", err)
		return
	}
	for _, part := range parts {
		fmt.Printf("part:%v\n", part)
		diskInfo, _ := disk.Usage(part.Mountpoint)
		fmt.Printf("disk info used:%v   free:%v\n", diskInfo.UsedPercent, diskInfo.Free)
	}
	ioStat, _ := disk.IOCounters()
	for k, v := range ioStat {
		fmt.Printf("%v:%v\n", k, v)
	}
}

7、net IO

func getNetInfo() {
	info, _ := net.IOCounters(true)
	for index, v := range info {
		fmt.Printf("%v:%v sent:%v recv:%v\n", index, v, v.BytesSent, v.BytesRecv)
	}
}

8、获取本机IP的两种方式

func getLocalIP() (ip string, err error) {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return
	}
	for _, addr := range addrs {
		ipAddr, ok := addr.(*net.IPNet)
		if !ok {
			continue
		}
		if ipAddr.IP.IsLoopback() {
			continue
		}
		if !ipAddr.IP.IsGlobalUnicast() {
			continue
		}
		return ipAddr.IP.String(), nil
	}
	return
}

func getOutboundIP() string {
	conn, err := net.Dial("udp", "8.8.8.8:80")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	localAddr := conn.LocalAddr().(*net.UDPAddr)
	fmt.Println(localAddr.String())
	return localAddr.IP.String()
}

五、influxDB

influxDB是一个开源分布式时序、事件和指标数据库。使用Go语言编写，无需外部依赖，其设计目标是实现分布式和水平伸缩扩展。常用的一种使用场景：监控数据统计。每毫秒记录一下电脑内存的使用情况，然后就可以根据统计的数据，利用图形化界面（InfluxDB V1一般配合Grafana）制作内存使用情况的折线图；

1、下载安装

下载地址：https://portal.influxdata.com/downloads/

这个网站引用了google的api，所以在国内点页面的按钮没反应，那how do we do？

通过检查元素，搜索。或者简单粗暴：https://dl.influxdata.com/influxdb/releases/influxdb-1.7.3_windows_amd64.zip