This is a rules engine for use with etcd. Simple dynamic rules allow the specificationof keys based on the gin attribute syntax and the comparison to literals or otherkeys. These rules can be nested inside of AND, OR and NOT rules to enable the expressionof complex relationships of values in etcd and the actions to be triggered when a setof conditions has been met. The engine watches etcd for updates and crawls the data treeat configurable intervals. This library makes use of the IBM-Cloud/go-etcd-lock libraryto enable concurrent monitoring by multiple application instances without collisions--thefirst client to obtain the lock processes the change while the others quickly fail to acquirethe lock and move on. A trigger callback function should update the model if the actionis successful so it is not retriggered. Recurring actions, such as continuous polling,can be implemented with rules that reference nodes with TTLs such that the expiration ofa node triggers a rule and the callback adds back a node with the same key and TTL.
package mainimport ("time""github.com/IBM-Cloud/go-etcd-rules/rules" v3"go.etcd.io/etcd/client/v3""github.com/uber-go/zap""golang.org/x/net/context")funclTP(valstring)*string {s:=valreturn&s}funcmain() {logger:=zap.New(zap.NewJSONEncoder(zap.RFC3339Formatter("ts")),zap.AddCaller(), )cfg:= v3.Config{Endpoints: []string{"http://127.0.0.1:2379"}, }engine:=rules.NewV3Engine(cfg,logger)serverActive,err:=rules.NewEqualsLiteralRule("/servers/:serverid/state",lTP("active"))iferr!=nil {panic(err) }pollDelayGone,err:=rules.NewEqualsLiteralRule("/servers/internal/:serverid/poll_delay",nil)iferr!=nil {panic(err) }engine.AddRule(rules.NewAndRule(serverActive,pollDelayGone),"/:serverid/poll",pollServer,RuleID("example") )engine.Run()end:=make(chanbool)<-end}funcpollServer(task*rules.V3RuleTask) {cl,err:=v3.New(task.Conf)iferr!=nil {return }kv:=v3.NewKV(cl)resp,err:=kv.Get(task.Context,task.Attr.Format("/servers/:serverid/ip"))iferr!=nil {return }ifresp.Count==0 {return }ip:=string(resp.Kvs[0].Value)varstatusstringifping(ip.Node.Value) {status="ok" }else {status="down" }kv.Put(task.Context,task.Attr.Format("/servers/:serverid/status"),status)// Add new poll delayleaseCtx,cancel:=context.WithTimeout(context.Background(),time.Second)defercancel()resp,err=cl.Grant(leaseCtx,int64(5))iferr!=nil {return }kv.Put(task.Context,task.Attr.Format("/servers/internal/:serverid/poll_delay"),"",v3.WithLease(resp.ID))}
