Merge pull request #134 from SkyAPM/develop

New Agent

Author: heyanlong

docs/zh/install-sdk.md

@@ -51,8 +51,8 @@ skywalking.enable=1
 skywalking.version=6
 ; app_code代码，不要含特殊字符，请使用数字、字母、下换线。(默认为：hello_skywalking)
 skywalking.app_code=hello_skywalking
-; sock文件路径（默认值为/tmp/sky_agent.sock）
-skywalking.sock_path=/tmp/sky_agent.sock
+; sock文件路径（默认值为/tmp/sky-agent.sock）
+skywalking.sock_path=/tmp/sky-agent.sock
 ```
 
 

skywalking.c

@@ -70,6 +70,7 @@ ZEND_DECLARE_MODULE_GLOBALS(skywalking)
 static int le_skywalking;
 static int application_instance = 0;
 static int application_id = 0;
+static char application_uuid[37] = {0};
 static int sky_increment_id = 0;
 static int cli_debug = 0;
 
@@ -85,7 +86,7 @@ PHP_INI_BEGIN()
 	STD_PHP_INI_BOOLEAN("skywalking.enable",   	"0", PHP_INI_ALL, OnUpdateBool, enable, zend_skywalking_globals, skywalking_globals)
 	STD_PHP_INI_ENTRY("skywalking.version",   	"6", PHP_INI_ALL, OnUpdateLong, version, zend_skywalking_globals, skywalking_globals)
 	STD_PHP_INI_ENTRY("skywalking.app_code", "hello_skywalking", PHP_INI_ALL, OnUpdateString, app_code, zend_skywalking_globals, skywalking_globals)
-	STD_PHP_INI_ENTRY("skywalking.sock_path", "/tmp/sky_agent.sock", PHP_INI_ALL, OnUpdateString, sock_path, zend_skywalking_globals, skywalking_globals)
+	STD_PHP_INI_ENTRY("skywalking.sock_path", "/tmp/sky-agent.sock", PHP_INI_ALL, OnUpdateString, sock_path, zend_skywalking_globals, skywalking_globals)
 PHP_INI_END()
 
 /* }}} */
@@ -1361,6 +1362,7 @@ static void request_init() {
     generate_context();
 
     add_assoc_long(&SKYWALKING_G(UpstreamSegment), "application_instance", application_instance);
+    add_assoc_stringl(&SKYWALKING_G(UpstreamSegment), "uuid", application_uuid, strlen(application_uuid));
     add_assoc_long(&SKYWALKING_G(UpstreamSegment), "pid", getppid());
     add_assoc_long(&SKYWALKING_G(UpstreamSegment), "application_id", application_id);
     add_assoc_long(&SKYWALKING_G(UpstreamSegment), "version", SKYWALKING_G(version));
@@ -1533,9 +1535,10 @@ static int sky_register() {
                     p = strtok(NULL, ",");
                 }
 
-                if (ids[0] != NULL && ids[1] != NULL) {
+                if (ids[0] != NULL && ids[1] != NULL && ids[2] != NULL) {
                     application_id = atoi(ids[0]);
                     application_instance = atoi(ids[1]);
+                    strncpy(application_uuid, ids[2], sizeof application_uuid - 1);
                 }
             }
 

src/agent/cmd/main.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"agent/agent/logger"
+	"agent/agent/service"
+	"github.com/urfave/cli"
+	"os"
+)
+
+var log = logger.Log
+
+func main() {
+
+	defer func() {
+		if err := recover(); err != nil {
+			log.Error(err)
+		}
+	}()
+
+	app := cli.NewApp()
+	app.Name = "sky_php_agent"
+	app.Usage = "the skywalking trace sending agent"
+	app.Flags = []cli.Flag{
+		cli.StringFlag{Name: "grpc", Usage: "SkyWalking collector grpc address", Value: "127.0.0.1:11800"},
+		cli.StringFlag{Name: "socket", Usage: "Pipeline for communicating with PHP", Value: "/tmp/sky-agent.sock"},
+		cli.IntFlag{Name: "send-rate", Usage: "Send trace 1 second by default", Value: 1},
+	}
+
+	app.Action = func(c *cli.Context) error {
+
+		a := service.NewAgent(c)
+		a.Run()
+		return nil
+	}
+
+	err := app.Run(os.Args)
+	if err != nil {
+		log.Errorln(err)
+	}
+}

src/agent/logger/log.go

@@ -0,0 +1,19 @@
+package logger
+
+import (
+	"github.com/sirupsen/logrus"
+	"os"
+)
+
+var Log *logrus.Logger
+
+func init() {
+	if Log == nil {
+		Log = logrus.New()
+	}
+	Log.SetOutput(os.Stdout)
+	Log.SetFormatter(&logrus.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: "2006-01-02 15:04:05",
+	})
+}

src/agent/service/agent.go

@@ -0,0 +1,153 @@
+package service
+
+import (
+	"agent/agent/logger"
+	"agent/agent/pb/agent"
+	"agent/agent/pb/agent2"
+	"agent/agent/pb/register2"
+	"container/list"
+	"fmt"
+	"github.com/urfave/cli"
+	"google.golang.org/grpc"
+	"net"
+	"os"
+	"sync"
+	"time"
+)
+
+var log = logger.Log
+
+type register struct {
+	c    net.Conn
+	body string
+}
+
+type grpcClient struct {
+	segmentClientV5 agent.TraceSegmentServiceClient
+	segmentClientV6 agent2.TraceSegmentReportServiceClient
+	pingClient5     agent.InstanceDiscoveryServiceClient
+	pintClient6     register2.ServiceInstancePingClient
+}
+
+type Agent struct {
+	flag              *cli.Context
+	grpcConn          *grpc.ClientConn
+	grpcClient        grpcClient
+	socket            string
+	socketListener    net.Listener
+	register          chan *register
+	registerCache     sync.Map
+	registerCacheLock sync.Mutex
+	trace             chan string
+	queue             *list.List
+}
+
+func NewAgent(cli *cli.Context) *Agent {
+	var agent = &Agent{
+		flag:     cli,
+		socket:   cli.String("socket"),
+		register: make(chan *register),
+		trace:    make(chan string),
+		queue:    list.New(),
+	}
+
+	go agent.sub()
+
+	return agent
+}
+
+func (t *Agent) Run() {
+	log.Info("hello skywalking")
+	t.connGRPC()
+	t.listenSocket()
+
+	defer func() {
+		var err error
+		err = t.socketListener.Close()
+		if err != nil {
+			log.Errorln(err)
+		}
+
+		err = t.grpcConn.Close()
+		if err != nil {
+			log.Errorln(err)
+		}
+	}()
+}
+
+func (t *Agent) connGRPC() {
+	var err error
+	grpcAdd := t.flag.String("grpc")
+	t.grpcConn, err = grpc.Dial(grpcAdd, grpc.WithInsecure())
+	if err != nil {
+		log.Panic(err)
+	}
+
+	log.Infof("connection %s...", grpcAdd)
+	t.grpcClient.segmentClientV5 = agent.NewTraceSegmentServiceClient(t.grpcConn)
+	t.grpcClient.segmentClientV6 = agent2.NewTraceSegmentReportServiceClient(t.grpcConn)
+	t.grpcClient.pingClient5 = agent.NewInstanceDiscoveryServiceClient(t.grpcConn)
+	t.grpcClient.pintClient6 = register2.NewServiceInstancePingClient(t.grpcConn)
+}
+
+func (t *Agent) listenSocket() {
+	var err error
+	if err = os.RemoveAll(t.socket); err != nil {
+		log.Panic(err)
+	}
+	t.socketListener, err = net.Listen("unix", t.socket)
+	if err != nil {
+		log.Panic(err)
+	}
+
+	err = os.Chmod(t.socket, os.ModeSocket|0777)
+	if err != nil {
+		log.Warningln(err)
+	}
+
+	for {
+		c, err := t.socketListener.Accept()
+		if err != nil {
+			log.Errorln(err)
+			break
+		}
+		// start a new goroutine to handle
+		// the new connection.
+		conn := NewConn(t, c)
+		go conn.Handle()
+	}
+}
+
+func (t *Agent) sub() {
+	heartbeatTicker := time.NewTicker(time.Second * 40)
+	defer heartbeatTicker.Stop()
+	traceSendTicker := time.NewTicker(time.Second * time.Duration(t.flag.Int("send-rate")))
+	defer traceSendTicker.Stop()
+
+	for {
+		select {
+		case <-traceSendTicker.C:
+			len := t.queue.Len()
+			if len > 0 {
+				var segments []*upstreamSegment
+				for i := 0; i < len; i++ {
+					// front top 100
+					e := t.queue.Front()
+					st := format(fmt.Sprintf("%v", e.Value))
+					if st != nil {
+						segments = append(segments, st)
+					}
+					t.queue.Remove(e)
+				}
+				go t.send(segments)
+			}
+		case <-heartbeatTicker.C:
+			go t.heartbeat()
+		case register := <-t.register:
+			go t.doRegister(register)
+		case trace := <-t.trace:
+			t.queue.PushBack(trace)
+			go t.recoverRegister(trace)
+		}
+	}
+}

src/agent/service/conn.go

@@ -0,0 +1,59 @@
+package service
+
+import (
+	"io"
+	"net"
+	"strings"
+)
+
+type Conn struct {
+	agent *Agent
+	c     net.Conn
+}
+
+func NewConn(a *Agent, c net.Conn) *Conn {
+	var conn = &Conn{
+		agent: a,
+		c:     c,
+	}
+
+	return conn
+}
+
+func (c *Conn) Handle() {
+
+	defer func() {
+		c.c.Close()
+	}()
+
+	buf := make([]byte, 4096)
+	var json string
+	var endIndex int
+	for {
+		n, err := c.c.Read(buf)
+		if err != nil {
+			if err != io.EOF {
+				log.Warn("conn read error:", err)
+			}
+			return
+		}
+		json += string(buf[0:n])
+		for {
+			endIndex = strings.IndexAny(json, "\n")
+			if endIndex >= 0 {
+				body := json[0:endIndex]
+				if body[:1] == "0" {
+					c.agent.register <- &register{
+						c:    c.c,
+						body: body[1:],
+					}
+				} else if body[:1] == "1" {
+					c.agent.trace <- body[1:]
+				}
+				json = json[endIndex+1:]
+			} else {
+				break
+			}
+		}
+	}
+}

src/agent/service/heartbeat.go

@@ -0,0 +1,48 @@
+package service
+
+import (
+	"agent/agent/pb/agent"
+	"agent/agent/pb/register2"
+	"context"
+	"time"
+)
+
+func (t *Agent) heartbeat() {
+
+	t.registerCache.Range(func(key, value interface{}) bool {
+		log.Infoln("heartbeat")
+		bind := value.(registerCache)
+
+		if bind.Version == 5 {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+			defer cancel()
+
+			_, err := t.grpcClient.pingClient5.Heartbeat(ctx, &agent.ApplicationInstanceHeartbeat{
+				ApplicationInstanceId: bind.InstanceId,
+				HeartbeatTime:         time.Now().UnixNano() / 1000000,
+			})
+
+			if err != nil {
+				log.Error("heartbeat:", err)
+			} else {
+				log.Infof("heartbeat appId %d appInsId %d", bind.AppId, bind.InstanceId)
+			}
+
+		} else if bind.Version == 6 {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+			defer cancel()
+
+			_, err := t.grpcClient.pintClient6.DoPing(ctx, &register2.ServiceInstancePingPkg{
+				ServiceInstanceId:   bind.InstanceId,
+				Time:                time.Now().UnixNano() / 1000000,
+				ServiceInstanceUUID: bind.Uuid,
+			})
+			if err != nil {
+				log.Error("heartbeat:", err)
+			} else {
+				log.Infof("heartbeat appId %d appInsId %d", bind.AppId, bind.InstanceId)
+			}
+		}
+		return true
+	})
+}