-
-
-
- OPEN
- JTS
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
Your open Juniper Telemetry Stack
+
+
+
+
+
+
+
+
+
+
+
+
+ Your open Juniper Telemetry Stack
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
DOWN
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
UP
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
- Empty DB
-
-
-
-
-
-
-
-
-
-
-
TELEGRAF VMX
+
+
+
+
+
+
+
+ Empty DB
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
{{.NumVMX}} Router(s) collected
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ {{.NumVMX}} Router(s) collected
+
-
-
-
+
+
+
+
+
+
-
-
-
-
-
-
@@ -70,7 +77,10 @@
Current profile associations
| {{.Shortname}} |
{{.Profiles}} |
- |
+ |
{{end}}
@@ -78,8 +88,8 @@ Current profile associations
-
+
diff --git a/main.go b/main.go
index be7f2b2..38760f0 100644
--- a/main.go
+++ b/main.go
@@ -9,6 +9,7 @@ import (
"jtso/kapacitor"
"jtso/logger"
_ "jtso/output"
+ _ "jtso/parser"
"jtso/portal"
"jtso/sqlite"
"jtso/worker"
diff --git a/parser/node.go b/parser/node.go
new file mode 100644
index 0000000..d67ede6
--- /dev/null
+++ b/parser/node.go
@@ -0,0 +1,118 @@
+package parser
+
+// TreeNode represents a node in the tree structure.
+type TreeNode struct {
+ Data interface{}
+ Value map[string]interface{}
+ Children []*TreeNode
+}
+
+// New creates a new instance of TreeNode with the given data and value.
+func NewTree(data, value interface{}) *TreeNode {
+ return &TreeNode{Data: data, Value: value.(map[string]interface{}), Children: make([]*TreeNode, 0)}
+}
+
+// AddChild adds a new child to the current node.
+func (n *TreeNode) AddChild(child *TreeNode) {
+ n.Children = append(n.Children, child)
+}
+
+// InsertChild inserts a new child into the tree structure.
+func (n *TreeNode) InsertChild(data interface{}, value map[string]interface{}) *TreeNode {
+ newChild := NewTree(data, value)
+ n.AddChild(newChild)
+ return newChild
+}
+
+// AddValue adds a value to the current node.
+func (n *TreeNode) AddValue(value map[string]interface{}) {
+ n.Value = mergeMap(n.Value, value)
+ //for key, value := range value {
+ // n.Value[key] = value
+ // }
+}
+
+func mergeMap(m1 map[string]interface{}, m2 map[string]interface{}) map[string]interface{} {
+ result := deepCopyMap(m1)
+ for k, v := range m2 {
+ found := false
+ if val, ok := result[k]; ok {
+ if _, isDictResult := val.(map[string]interface{}); isDictResult {
+ if _, isDictValue := v.(map[string]interface{}); isDictValue {
+ found = true
+ result[k] = mergeMap(result[k].(map[string]interface{}), v.(map[string]interface{}))
+ }
+ }
+ }
+ if !found {
+ result[k] = v
+ }
+ }
+ return result
+}
+
+func deepCopyMap(original map[string]interface{}) map[string]interface{} {
+ copyMap := make(map[string]interface{})
+
+ for key, value := range original {
+ switch v := value.(type) {
+ case map[string]interface{}:
+ // Recursively deep copy nested maps
+ copyMap[key] = deepCopyMap(v)
+ case []interface{}:
+ // Recursively deep copy nested slices
+ copyMap[key] = deepCopySlice(v)
+ default:
+ // Copy other types directly
+ copyMap[key] = value
+ }
+ }
+
+ return copyMap
+}
+
+func deepCopySlice(original []interface{}) []interface{} {
+ copySlice := make([]interface{}, len(original))
+
+ for i, v := range original {
+ switch element := v.(type) {
+ case map[string]interface{}:
+ // Recursively deep copy nested maps
+ copySlice[i] = deepCopyMap(element)
+ case []interface{}:
+ // Recursively deep copy nested slices
+ copySlice[i] = deepCopySlice(element)
+ default:
+ // Copy other types directly
+ copySlice[i] = element
+ }
+ }
+
+ return copySlice
+}
+
+// FindNode finds a node in the tree structure by searching for a specific piece of data.
+func (n *TreeNode) FindNode(targetData interface{}) (*TreeNode, bool) {
+ if n == nil {
+ return nil, false
+ }
+
+ if n.Data == targetData {
+ return n, true
+ }
+
+ for _, child := range n.Children {
+ if child.Data == targetData {
+ return child, true
+ }
+ }
+
+ return nil, false
+}
+
+func (n *TreeNode) Traverse(f func(node *TreeNode)) {
+ f(n)
+ for _, child := range n.Children {
+ child.Traverse(f)
+ }
+}
diff --git a/parser/parser.go b/parser/parser.go
new file mode 100644
index 0000000..1f21c33
--- /dev/null
+++ b/parser/parser.go
@@ -0,0 +1,389 @@
+package parser
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "jtso/logger"
+ "jtso/sqlite"
+ "net/http"
+ "reflect"
+ "regexp"
+ "strings"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/openconfig/gnmic/pkg/api"
+ "github.com/openconfig/gnmic/pkg/api/target"
+ "github.com/openconfig/gnmic/pkg/formatters"
+)
+
+const PATH_CERT string = "/var/shared/telegraf/cert/"
+
+var root *TreeNode
+var global []string
+var re1, re2 *regexp.Regexp
+var StreamObj *Streamer
+
+type TreeJs struct {
+ Id string `json:"id"`
+ Parent string `json:"parent"`
+ Text string `json:"text"`
+ Icon string `json:"icon"`
+}
+
+type Streamer struct {
+ Stream int
+ Path string
+ Router string
+ Port int
+ Merger bool
+ Ticker time.Time
+ ForceFlush bool
+ Result *TreeNode
+ Flusher http.Flusher
+ Writer http.ResponseWriter
+ Error error
+ StopStreaming chan struct{}
+}
+
+func genUUID() string {
+ return uuid.New().String()
+}
+
+func init() {
+ // init re
+ re1 = regexp.MustCompile("(\\d+)")
+ re2 = regexp.MustCompile("(.*)\\[(.*)=(.*)\\]")
+
+ // init streamer
+ StreamObj = new(Streamer)
+}
+
+func ToJSON(data map[string]interface{}) string {
+ // Convert the data map to JSON
+ jsonData, err := json.Marshal(data)
+ if err != nil {
+ fmt.Println("Error encoding JSON:", err)
+ return ""
+ }
+ return string(jsonData)
+}
+
+func StreamData(m string, s string, payload ...string) {
+ var pl string
+ if len(payload) == 0 {
+ pl = ""
+ } else {
+ pl = payload[0]
+ }
+
+ data := map[string]interface{}{
+ "msg": m,
+ "status": s,
+ "payload": pl,
+ }
+ jsonData := fmt.Sprintf("data: %s\n\n", ToJSON(data))
+ fmt.Fprint(StreamObj.Writer, jsonData)
+
+ // Compute the time between 2 flushs - min must be 1 sec
+ elapsedTime := time.Since(StreamObj.Ticker)
+ if elapsedTime.Seconds() >= 1.0 || StreamObj.ForceFlush {
+ StreamObj.Flusher.Flush()
+ }
+ StreamObj.Ticker = time.Now()
+
+}
+
+func advancedSplit(path string) []string {
+
+ if strings.Contains(path, "=") && strings.Contains(path, "[") {
+ var newPath string
+ escape := false
+
+ for _, w := range path {
+ if w == '[' {
+ escape = true
+ }
+ if w == ']' {
+ escape = false
+ }
+ if !escape {
+ if w == '/' {
+ newPath += "£££"
+ } else {
+ newPath += string(w)
+ }
+ } else {
+ newPath += string(w)
+ }
+ }
+ return strings.Split(newPath, "£££")
+ }
+ return strings.Split(path, "/")
+}
+
+func PrintTree(node map[string]interface{}, indent int, o map[string]interface{}, parentKey string, j *[]TreeJs) {
+ var entry TreeJs
+
+ for k, v := range node {
+ if reflect.TypeOf(v).Kind() == reflect.Map {
+ newkey := genUUID()
+ entry = TreeJs{
+ Id: newkey,
+ Parent: parentKey,
+ Text: k,
+ Icon: "fas fa-search-plus",
+ }
+ *j = append(*j, entry)
+
+ //fmt.Printf("%s+ %s\n", strings.Repeat(" ", indent), k)
+ o[k] = map[string]interface{}{}
+ PrintTree(v.(map[string]interface{}), indent+1, o[k].(map[string]interface{}), newkey, j)
+ } else {
+ o[k] = v
+ //fmt.Printf("%s+ %s: %s\n", strings.Repeat(" ", indent), k, fmt.Sprint(v))
+ entry = TreeJs{
+ Id: genUUID(),
+ Parent: parentKey,
+ Text: fmt.Sprintf("%s = %s", k, fmt.Sprint(v)),
+ Icon: "fas fa-sign-out-alt",
+ }
+ *j = append(*j, entry)
+ }
+ }
+
+}
+
+func TraverseTree(node *TreeNode, parentKey string, j *[]TreeJs) {
+ global = append(global, node.Data.(string))
+
+ if len(node.Children) != 0 {
+ for _, child := range node.Children {
+ TraverseTree(child, parentKey, j)
+ }
+ global = global[:len(global)-1]
+ } else {
+ path := strings.Join(global, "/")
+ var entry TreeJs
+ newkey := genUUID()
+
+ entry = TreeJs{
+ Id: newkey,
+ Parent: parentKey,
+ Text: path,
+ Icon: "fas fa-search-plus",
+ }
+ *j = append(*j, entry)
+
+ //fmt.Printf("%s\n", path)
+ output := make(map[string]interface{})
+ output[path] = make(map[string]interface{})
+ PrintTree(node.Value, 1, output[path].(map[string]interface{}), newkey, j)
+ global = global[:len(global)-1]
+ }
+}
+
+func parseXpath(xpath string, value string, merge bool) error {
+
+ var parent *TreeNode
+ var key []string
+ var val map[string]interface{}
+
+ key = make([]string, 0)
+
+ if merge {
+ xpath = re1.ReplaceAllString(xpath, "x")
+ }
+ StreamData(fmt.Sprintf("XPATH Extracted: %s", xpath), "OK")
+ lpath := advancedSplit(xpath)
+
+ parent = root
+ for i, v := range lpath {
+ if i == len(lpath)-1 {
+ if len(key) == 0 {
+ val["alone"] = value
+ } else {
+ val = make(map[string]interface{})
+ tmp := val
+ for ki, kv := range key {
+ if ki == len(key)-1 {
+ tmp[kv] = value
+ } else {
+ tmp[kv] = make(map[string]interface{})
+ tmp = tmp[kv].(map[string]interface{})
+ }
+ }
+ }
+ } else {
+ val = make(map[string]interface{})
+ }
+ if strings.Contains(v, "=") {
+ matches := re2.FindStringSubmatch(v)
+
+ composite := matches[1] + "[" + matches[2] + "=*]"
+ node, result := parent.FindNode(composite)
+ if result {
+ node.AddValue(val)
+ } else {
+ node = parent.InsertChild(composite, val)
+ }
+ parent = node
+ key = append(key, matches[3])
+ } else {
+ node, result := parent.FindNode(v)
+ if result {
+ node.AddValue(val)
+ } else {
+ node = parent.InsertChild(v, val)
+ }
+ parent = node
+ }
+ }
+ return nil
+}
+
+func LaunchSearch() {
+
+ logger.Log.Infof("Start subscription for router %s and xpath %s", StreamObj.Router, StreamObj.Path)
+ StreamData(fmt.Sprintf("Start subscription for router %s and xpath %s", StreamObj.Router, StreamObj.Path), "OK")
+
+ // Init global variable
+ root = NewTree("", map[string]interface{}{})
+ global = make([]string, 0)
+ var tg *target.Target
+ var err error
+
+ // Retrieve cred info
+ tls := false
+ skip := false
+ clienttls := false
+ if sqlite.ActiveCred.UseTls == "yes" {
+ tls = true
+ }
+ if sqlite.ActiveCred.SkipVerify == "yes" {
+ skip = true
+ }
+ if sqlite.ActiveCred.ClientTls == "yes" {
+ clienttls = true
+ }
+
+ if tls {
+ if clienttls {
+ StreamData("Try to create Target - option 3", "OK")
+ // create a target
+ tg, err = api.NewTarget(
+ api.Name("jtso"),
+ api.Address(StreamObj.Router+":"+fmt.Sprint(StreamObj.Port)),
+ api.Username(sqlite.ActiveCred.GnmiUser),
+ api.Password(sqlite.ActiveCred.GnmiPwd),
+ api.SkipVerify(skip),
+ api.Insecure(false),
+ api.TLSCA(PATH_CERT+"RootCA.crt"),
+ api.TLSCert(PATH_CERT+"client.crt"),
+ api.TLSKey(PATH_CERT+"client.key"),
+ )
+
+ } else {
+ StreamData("Try to create Target - option 2", "OK")
+ // create a target
+ tg, err = api.NewTarget(
+ api.Name("jtso"),
+ api.Address(StreamObj.Router+":"+fmt.Sprint(StreamObj.Port)),
+ api.Username(sqlite.ActiveCred.GnmiUser),
+ api.Password(sqlite.ActiveCred.GnmiPwd),
+ api.SkipVerify(skip),
+ api.Insecure(false),
+ api.TLSCA(PATH_CERT+"RootCA.crt"),
+ )
+
+ }
+ } else {
+ StreamData("Try to create Target - option 1", "OK")
+ // create a target
+ tg, err = api.NewTarget(
+ api.Name("jtso"),
+ api.Address(StreamObj.Router+":"+fmt.Sprint(StreamObj.Port)),
+ api.Username(sqlite.ActiveCred.GnmiUser),
+ api.Password(sqlite.ActiveCred.GnmiPwd),
+ api.SkipVerify(skip),
+ api.Insecure(true),
+ )
+ }
+
+ if err != nil {
+ logger.Log.Errorf("Unable to create gNMI target: %v", err)
+ StreamData(fmt.Sprintf("Unable to create gNMI target: %v", err), "ERROR")
+ StreamObj.Error = err
+ close(StreamObj.StopStreaming)
+ return
+ }
+ StreamData("Create gNMI Target", "OK")
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ err = tg.CreateGNMIClient(ctx)
+ if err != nil {
+ logger.Log.Errorf("Unable to create gNMI client: %v", err)
+ StreamData(fmt.Sprintf("Unable to create gNMI client: %v", err), "ERROR")
+ StreamObj.Error = err
+ close(StreamObj.StopStreaming)
+ return
+ }
+ StreamData("Create gNMI Client", "OK")
+
+ defer tg.Close()
+ // create a gNMI subscribeRequest
+ subReq, err := api.NewSubscribeRequest(
+ api.Encoding("proto"),
+ api.SubscriptionListMode("stream"),
+ api.Subscription(
+ api.Path(StreamObj.Path),
+ api.SubscriptionMode("sample"),
+ api.SampleInterval(30*time.Second),
+ ))
+ if err != nil {
+ logger.Log.Errorf("Unable to create gNMI subscription: %v", err)
+ StreamData(fmt.Sprintf("Unable to create gNMI subscription: %v", err), "ERROR")
+ StreamObj.Error = err
+ close(StreamObj.StopStreaming)
+ return
+ }
+ StreamData("Create gNMI Subscription", "OK")
+
+ go tg.Subscribe(ctx, subReq, "sub1")
+
+ go func() {
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(40 * time.Second):
+ logger.Log.Infof("End of the subscription timer")
+ tg.StopSubscription("sub1")
+ }
+ }()
+
+ subRspChan, subErrChan := tg.ReadSubscriptions()
+ StreamData("Start collection data", "OK")
+ StreamObj.ForceFlush = false
+ for {
+ select {
+ case rsp := <-subRspChan:
+ r, _ := formatters.ResponsesFlat(rsp.Response)
+ for k, v := range r {
+ parseXpath(k, fmt.Sprint(v), StreamObj.Merger)
+ }
+
+ case gnmiErr := <-subErrChan:
+ //traverseTree(root)
+ StreamObj.ForceFlush = true
+ logger.Log.Infof("End of the subscription after the 40 secs analysis - status of the end: %v", gnmiErr.Err.Error())
+ StreamObj.Error = gnmiErr.Err
+ time.Sleep(1 * time.Second)
+ StreamObj.Result = root
+ close(StreamObj.StopStreaming)
+ return
+ }
+ }
+
+}
diff --git a/portal/msg.go b/portal/msg.go
index d55089a..aef3575 100644
--- a/portal/msg.go
+++ b/portal/msg.go
@@ -11,6 +11,12 @@ type (
Shortname string `json:"shortname"`
}
+ SearchPath struct {
+ Shortname string `json:"shortname"`
+ Xpath string `json:"xpath"`
+ Merge bool `json:"merge"`
+ }
+
RouterDetails struct {
Hostname string `json:"hostname"`
Shortname string `json:"shortname"`
diff --git a/portal/webapp.go b/portal/webapp.go
index a3c8b5e..01c415e 100644
--- a/portal/webapp.go
+++ b/portal/webapp.go
@@ -2,6 +2,8 @@ package portal
import (
"context"
+ "encoding/json"
+ "fmt"
"html/template"
"io"
"jtso/association"
@@ -9,11 +11,13 @@ import (
"jtso/influx"
"jtso/logger"
"jtso/netconf"
+ "jtso/parser"
"jtso/sqlite"
"jtso/worker"
"net/http"
"strconv"
"strings"
+ "time"
"github.com/labstack/echo/v4"
"github.com/labstack/echo/v4/middleware"
@@ -54,6 +58,7 @@ func New(cfg *config.ConfigContainer) *WebApp {
wapp := echo.New()
//configure app
wapp.Use(middleware.Static("html/assets"))
+ wapp.Use(middleware.CORS())
//Templating config
wapp.Renderer = &TemplateRegistry{
@@ -67,6 +72,8 @@ func New(cfg *config.ConfigContainer) *WebApp {
wapp.GET("/profiles.html", routeProfiles)
wapp.GET("/cred.html", routeCred)
wapp.GET("/doc.html", routeDoc)
+ wapp.GET("/browser.html", routeBrowse)
+ wapp.GET("/stream", routeStream)
// configure POST routers
wapp.POST("/addrouter", routeAddRouter)
@@ -76,6 +83,7 @@ func New(cfg *config.ConfigContainer) *WebApp {
wapp.POST("/updatecred", routeUptCred)
wapp.POST("/updatedoc", routeUptDoc)
wapp.POST("/influxmgt", routeInfluxMgt)
+ wapp.POST("/searchxpath", routeSearchPath)
collectCfg = new(collectInfo)
collectCfg.cfg = cfg
@@ -85,6 +93,7 @@ func New(cfg *config.ConfigContainer) *WebApp {
listen: ":" + strconv.Itoa(cfg.Portal.Port),
app: wapp,
}
+
}
func (w *WebApp) Run() {
@@ -250,6 +259,20 @@ func routeDoc(c echo.Context) error {
return c.Render(http.StatusOK, "doc.html", map[string]interface{}{"Profiles": lp, "GrafanaPort": grafanaPort})
}
+func routeBrowse(c echo.Context) error {
+ grafanaPort := collectCfg.cfg.Grafana.Port
+
+ // Get all routers from db
+ var lr []RouterDetails
+ lr = make([]RouterDetails, 0)
+
+ for _, r := range sqlite.RtrList {
+ lr = append(lr, RouterDetails{Hostname: r.Hostname, Shortname: r.Shortname, Family: r.Family, Model: r.Model, Version: r.Version})
+ }
+
+ return c.Render(http.StatusOK, "browser.html", map[string]interface{}{"Rtrs": lr, "GrafanaPort": grafanaPort})
+}
+
func routeAddRouter(c echo.Context) error {
var err error
@@ -403,6 +426,106 @@ func routeAddProfile(c echo.Context) error {
}
+func routeSearchPath(c echo.Context) error {
+ var err error
+
+ // check if other instance is already running
+ if parser.StreamObj.Stream != 0 {
+ logger.Log.Errorf("Streaming already running for path %s", parser.StreamObj.Path)
+ return c.JSON(http.StatusOK, Reply{Status: "NOK", Msg: "Another instance is currently requesting XPATH search. Retry later..."})
+ }
+ // change the streamer state to pending stream API request
+ parser.StreamObj.Stream = 1
+
+ r := new(SearchPath)
+ err = c.Bind(r)
+
+ if err != nil {
+ logger.Log.Errorf("Unable to parse Post request for searching XPATH: %v", err)
+ return c.JSON(http.StatusOK, Reply{Status: "NOK", Msg: "Unable to parse Post request for searching XPATH"})
+ }
+ h := ""
+ for _, i := range sqlite.RtrList {
+ if i.Shortname == r.Shortname {
+ h = i.Hostname
+ break
+ }
+ }
+ parser.StreamObj.Router = h
+ parser.StreamObj.Port = collectCfg.cfg.Gnmi.Port
+ parser.StreamObj.Path = r.Xpath
+ parser.StreamObj.Merger = r.Merge
+ parser.StreamObj.StopStreaming = make(chan struct{})
+
+ return c.JSON(http.StatusOK, Reply{Status: "OK", Msg: "Streaming well started."})
+}
+
+func routeStream(c echo.Context) error {
+ // Set the response header for Server-Sent Events
+ c.Response().Header().Set(echo.HeaderContentType, "text/event-stream")
+ c.Response().Header().Set("Cache-Control", "no-cache")
+ c.Response().Header().Set("Connection", "keep-alive")
+
+ // Flush the response buffer
+ c.Response().Flush()
+
+ if parser.StreamObj.Stream == 0 {
+ logger.Log.Errorf("Bad request - direct access of /stream is not allowed")
+ c.JSON(http.StatusBadRequest, Reply{Status: "NOK", Msg: "Bad request - direct access of /stream is not allowed"})
+ c.Response().Flush()
+ return nil
+ } else if parser.StreamObj.Stream == 1 {
+
+ // change the state of the stream to streaming
+ parser.StreamObj.Stream = 2
+ // Pass the context to parser
+ parser.StreamObj.Flusher, _ = c.Response().Writer.(http.Flusher)
+ parser.StreamObj.Writer = c.Response().Writer
+ parser.StreamObj.Ticker = time.Now()
+ parser.StreamObj.ForceFlush = true
+ // launch parser
+ go parser.LaunchSearch()
+ // loop until the end
+ for {
+ select {
+ case <-parser.StreamObj.StopStreaming:
+ var jsTree []parser.TreeJs
+ // depending on the error report:
+ errString := parser.StreamObj.Error.Error()
+
+ // Normal end
+ if strings.Contains(errString, "context canceled") {
+
+ parser.StreamData("End of the subscription. Close gNMI session", "OK")
+ logger.Log.Info("Generate payload based on the Tree")
+ jsTree = make([]parser.TreeJs, 0)
+ parser.TraverseTree(parser.StreamObj.Result, "#", &jsTree)
+ jsonData, err := json.Marshal(jsTree)
+ if err != nil {
+ logger.Log.Errorf("Unable to marshall the result: %v", err)
+ parser.StreamData(fmt.Sprintf("Unable to marshall the result: %s", err.Error()), "ERROR")
+ } else {
+ logger.Log.Info("Marshall the result: success")
+ // Convert the JSON data to a string
+ jsonString := string(jsonData)
+ parser.StreamData("End of the collection.", "END", jsonString)
+ }
+ } else {
+ logger.Log.Errorf("Unexpected gnmi error: %v", errString)
+ parser.StreamData(fmt.Sprintf("Unexpected gnmi error: %s", errString), "ERROR")
+ }
+
+ parser.StreamObj.Stream = 0
+ logger.Log.Info("Streaming has been now stopped properly...")
+ time.Sleep(500 * time.Millisecond)
+ return nil
+ }
+ }
+ }
+ return nil
+
+}
+
func routeDelProfile(c echo.Context) error {
var err error