don't block when loading identites any more. let callbacks drive it back to the UI

This commit is contained in:
dovholuknf
2021-01-20 15:54:10 -05:00
parent df3fa5eb08
commit aaa7334a0d
4 changed files with 116 additions and 114 deletions
+67 -76
View File
@@ -26,7 +26,7 @@ package cziti
#include "ziti/ziti_log.h"
#include "sdk.h"
extern void initCB(ziti_context nf, int status, void *ctx);
extern void zitiContextEvent(ziti_context nf, int status, void *ctx);
extern void eventCB(ziti_context ztx, ziti_event_t *event);
extern void serviceCB(ziti_context nf, ziti_service*, int status, void *ctx);
@@ -96,23 +96,25 @@ type ZService struct {
}
type ZIdentity struct {
Options *C.ziti_options
zctx C.ziti_context
zid *C.ziti_identity
status int
statusErr error
Loaded bool
Name string
Version string
Services sync.Map
Fingerprint string
Active bool
Options *C.ziti_options
czctx C.ziti_context
czid *C.ziti_identity
status int
statusErr error
Loaded bool
Name string
Version string
Services sync.Map
Fingerprint string
Active bool
StatusChanges chan int
}
func NewZid() *ZIdentity {
zid := &ZIdentity{}
zid.Services = sync.Map{}
zid.Options = (*C.ziti_options)(C.calloc(1, C.sizeof_ziti_options))
zid.StatusChanges = make(chan int)
return zid
}
@@ -120,18 +122,18 @@ func (c *ZIdentity) GetMetrics() (int64, int64, bool) {
if c == nil {
return 0, 0, false
}
if C.is_null(unsafe.Pointer(c.zctx)) {
if C.is_null(unsafe.Pointer(c.czctx)) {
log.Warnf("ziti context is C.NULL! %s", c.Fingerprint)
return 0, 0, false
}
var up, down C.double
C.ziti_get_transfer_rates(c.zctx, &up, &down)
C.ziti_get_transfer_rates(c.czctx, &up, &down)
return int64(up), int64(down), true
}
func (c *ZIdentity) UnsafePointer() unsafe.Pointer {
return unsafe.Pointer(c.zctx)
return unsafe.Pointer(c.czctx)
}
func (c *ZIdentity) AsKey() string {
return "marker askey"
@@ -147,8 +149,8 @@ func (c *ZIdentity) setVersionFromId() string {
}
c.Version = "<unknown version>"
if c != nil {
if c.zctx != nil {
v1 := C.ziti_get_controller_version(c.zctx)
if c.czctx != nil {
v1 := C.ziti_get_controller_version(c.czctx)
return C.GoString(v1.version)
}
}
@@ -161,9 +163,9 @@ func (c *ZIdentity) setNameFromId() string {
}
c.Name = "<unknown>"
if c != nil {
if c.zid != nil {
if c.zid.name != nil {
c.Name = C.GoString(c.zid.name)
if c.czid != nil {
if c.czid.name != nil {
c.Name = C.GoString(c.czid.name)
} else {
log.Debug("in Name - c.zid.name was nil")
}
@@ -177,8 +179,8 @@ func (c *ZIdentity) setNameFromId() string {
}
func (c *ZIdentity) Tags() []string {
if c.zctx != nil && c.zid != nil {
C.c_mapiter(&c.zid.tags)
if c.czctx != nil && c.czid != nil {
C.c_mapiter(&c.czid.tags)
/*
it := C.model_map_iterator(&c.zid.tags)
for {
@@ -197,8 +199,8 @@ func (c *ZIdentity) Tags() []string {
}
func (c *ZIdentity) Controller() string {
if c.zctx != nil {
return C.GoString(C.ziti_get_controller(c.zctx))
if c.czctx != nil {
return C.GoString(C.ziti_get_controller(c.czctx))
}
return C.GoString(c.Options.controller)
}
@@ -242,7 +244,7 @@ func serviceCB(_ C.ziti_context, service *C.ziti_service, status C.int, tnlr_ctx
var v1Config C.ziti_server_cfg_v1
r := C.ziti_service_get_config(service, cTunServerCfgName, unsafe.Pointer(&v1Config), (*[0]byte)(C.parse_ziti_server_cfg_v1))
if r == 0 {
C.ziti_tunneler_host_v1(C.tunneler_context(theTun.tunCtx), unsafe.Pointer(zid.zctx), service.name, v1Config.protocol, v1Config.hostname, v1Config.port)
C.ziti_tunneler_host_v1(C.tunneler_context(theTun.tunCtx), unsafe.Pointer(zid.czctx), service.name, v1Config.protocol, v1Config.hostname, v1Config.port)
C.free_ziti_server_cfg_v1(&v1Config)
} else {
log.Infof("service is bindable but doesn't have config? %s. flags: %v.", name, service.perm_flags)
@@ -267,10 +269,10 @@ func serviceCB(_ C.ziti_context, service *C.ziti_service, status C.int, tnlr_ctx
if err != nil {
log.Warn(err)
log.Infof("service intercept beginning for service: %s@%s:%d on ip %s", name, host, port, ip.String())
AddIntercept(svcId, name, ip.String(), port, unsafe.Pointer(zid.zctx))
AddIntercept(svcId, name, ip.String(), port, unsafe.Pointer(zid.czctx))
} else {
log.Infof("service intercept beginning for service: %s@%s:%d on ip %s", name, host, port, ip.String())
AddIntercept(svcId, name, ip.String(), port, unsafe.Pointer(zid.zctx))
AddIntercept(svcId, name, ip.String(), port, unsafe.Pointer(zid.czctx))
}
added := ZService{
Name: name,
@@ -314,13 +316,12 @@ func serviceUnavailable(ctx *ZIdentity, svcId string, name string) {
//export eventCB
func eventCB(ztx C.ziti_context, event *C.ziti_event_t) {
appCtx := C.ziti_app_ctx(ztx)
log.Debugf("events received. type: %d for ztx(%p)", event._type, ztx)
switch event._type {
case C.ZitiContextEvent:
ctxEvent := C.ziti_event_context_event(event)
initCB(ztx, ctxEvent.ctrl_status, appCtx)
zitiContextEvent(ztx, ctxEvent.ctrl_status, appCtx)
case C.ZitiRouterEvent:
rtrEvent := C.ziti_event_router_event(event)
@@ -368,38 +369,30 @@ func eventCB(ztx C.ziti_context, event *C.ziti_event_t) {
}
}
//export initCB
func initCB(nf C.ziti_context, status C.int, data unsafe.Pointer) {
ctx := (*ZIdentity)(data)
//export zitiContextEvent
func zitiContextEvent(nf C.ziti_context, status C.int, data unsafe.Pointer) {
zid := (*ZIdentity)(data)
ctx.status = int(status)
ctx.statusErr = zitiError(status)
ctx.zctx = nf
zid.status = int(status)
zid.statusErr = zitiError(status)
zid.czctx = nf
cfg := C.GoString(ctx.Options.config)
cfg := C.GoString(zid.Options.config)
if status == C.int(0) {
if nf != nil {
ctx.zid = C.ziti_get_identity(nf)
zid.czid = C.ziti_get_identity(nf)
}
ctx.Name = ctx.setNameFromId()
ctx.Version = ctx.setVersionFromId()
log.Infof("connected to controller %s running %v", ctx.Name, ctx.Version)
zid.Name = zid.setNameFromId()
zid.Version = zid.setVersionFromId()
log.Infof("============ controller connected: %s at %v", zid.Name, zid.Version)
} else {
log.Infof("failed to connect[%s] to controller for %s", ctx.statusErr, cfg)
}
ch, firstTime := initMap[cfg]
if firstTime {
ch <- ctx
delete(initMap, cfg)
log.Errorf("zitiContextEvent failed to connect[%s] to controller for %s", zid.statusErr, cfg)
}
zid.StatusChanges <- int(status)
}
var initMap = make(map[string]chan *ZIdentity)
func zitiError(code C.int) error {
if int(code) != 0 {
return errors.New(C.GoString(C.ziti_errorstr(code)))
@@ -408,37 +401,35 @@ func zitiError(code C.int) error {
}
func LoadZiti(cfg string, isActive bool) *ZIdentity {
ctx := NewZid()
zid := NewZid()
ctx.Active = isActive
zid.Active = isActive
ctx.Options.config = C.CString(cfg)
ctx.Options.refresh_interval = C.long(300)
ctx.Options.metrics_type = C.INSTANT
ctx.Options.config_types = C.all_configs
ctx.Options.pq_domain_cb = C.ziti_pq_domain_cb(C.ziti_pq_domain_go)
ctx.Options.pq_mac_cb = C.ziti_pq_mac_cb(C.ziti_pq_mac_go)
ctx.Options.pq_os_cb = C.ziti_pq_os_cb(C.ziti_pq_os_go)
ctx.Options.pq_process_cb = C.ziti_pq_process_cb(C.ziti_pq_process_go)
zid.Options.config = C.CString(cfg)
zid.Options.refresh_interval = C.long(300)
zid.Options.metrics_type = C.INSTANT
zid.Options.config_types = C.all_configs
zid.Options.pq_domain_cb = C.ziti_pq_domain_cb(C.ziti_pq_domain_go)
zid.Options.pq_mac_cb = C.ziti_pq_mac_cb(C.ziti_pq_mac_go)
zid.Options.pq_os_cb = C.ziti_pq_os_cb(C.ziti_pq_os_go)
zid.Options.pq_process_cb = C.ziti_pq_process_cb(C.ziti_pq_process_go)
ctx.Options.events = C.ZitiContextEvent | C.ZitiServiceEvent | C.ZitiRouterEvent
ctx.Options.event_cb = C.ziti_event_cb(C.eventCB)
ctx.Options.app_ctx = unsafe.Pointer(ctx)
zid.Options.events = C.ZitiContextEvent | C.ZitiServiceEvent | C.ZitiRouterEvent
zid.Options.event_cb = C.ziti_event_cb(C.eventCB)
ptr := unsafe.Pointer(zid)
zid.Options.app_ctx = ptr
ch := make(chan *ZIdentity)
initMap[cfg] = ch
rc := C.ziti_init_opts(ctx.Options, _impl.libuvCtx.l)
if rc != C.ZITI_OK {
ctx.status, ctx.statusErr = int(rc), zitiError(rc)
go func() {
ch <- ctx
}()
}
go func() {
rc := C.ziti_init_opts(zid.Options, _impl.libuvCtx.l)
if rc != C.ZITI_OK {
zid.status, zid.statusErr = int(rc), zitiError(rc)
log.Errorf("FAILED to load identity from config file: %s due to: %s", cfg, zid.statusErr)
} else {
log.Debugf("successfully loaded identity from config file: %s", cfg)
}
}()
res := <-ch
delete(initMap, cfg)
return res
return zid
}
//export free_async
+1 -1
View File
@@ -24,7 +24,7 @@ import (
func dbg() {
r := rts.ToStatus()
r := rts.ToStatus(true)
events.broadcast <- dto.TunnelStatusEvent{
StatusEvent: dto.StatusEvent{Op: "status"},
Status: r,
+9 -10
View File
@@ -77,9 +77,6 @@ func SubMain(ops chan string, changes chan<- svc.Status) error {
// a channel to signal the handleEvents that initialization is complete
initialized := make(chan struct{})
// setup events handler - needs to be before initialization
go handleEvents(initialized)
// initialize the network interface
err := initialize(cLogLevel)
if err != nil {
@@ -89,6 +86,8 @@ func SubMain(ops chan string, changes chan<- svc.Status) error {
setTunnelState(true)
go handleEvents(initialized)
//listen for services that show up
go acceptServices()
@@ -539,7 +538,7 @@ func serveEvents(conn net.Conn) {
log.Info("new event client connected - sending current status")
err := o.Encode( dto.TunnelStatusEvent{
StatusEvent: dto.StatusEvent{Op: "status"},
Status: rts.ToStatus(),
Status: rts.ToStatus(true),
ApiVersion: API_VERSION,
})
@@ -567,7 +566,7 @@ loop:
}
func reportStatus(out *json.Encoder) {
s := rts.ToStatus()
s := rts.ToStatus(true)
respond(out, dto.ZitiTunnelStatus{
Status: &s,
Metrics: nil,
@@ -776,13 +775,13 @@ func connectIdentity(id *Id) {
id.Services = append(id.Services, nil)
return true
})
}
events.broadcast <- dto.IdentityEvent{
ActionEvent: IDENTITY_ADDED,
Id: id.Identity,
events.broadcast <- dto.IdentityEvent{
ActionEvent: IDENTITY_ADDED,
Id: id.Identity,
}
log.Infof("connecting identity completed: %s[%s]", id.Name, id.FingerPrint)
}
log.Infof("connecting identity completed: %s[%s]", id.Name, id.FingerPrint)
}
func disconnectIdentity(id *Id) error {
+39 -27
View File
@@ -66,6 +66,7 @@ func (t *RuntimeState) SaveState() {
}
cfg, err := os.OpenFile(config.File(), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
defer cfg.Close()
if err != nil {
log.Panicf("An unexpected and unrecoverable error has occurred while %s: %v", "opening the config file", err)
}
@@ -73,7 +74,7 @@ func (t *RuntimeState) SaveState() {
w := bufio.NewWriter(bufio.NewWriter(cfg))
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
_ = enc.Encode(t.ToStatus())
_ = enc.Encode(t.ToStatus(false))
_ = w.Flush()
err = cfg.Close()
@@ -103,7 +104,7 @@ func backupConfig() (string, error) {
return backup, err
}
func (t *RuntimeState) ToStatus() dto.TunnelStatus {
func (t *RuntimeState) ToStatus(onlyInitialized bool) dto.TunnelStatus {
var uptime int64
now := time.Now()
@@ -113,7 +114,7 @@ func (t *RuntimeState) ToStatus() dto.TunnelStatus {
clean := dto.TunnelStatus{
Active: t.state.Active,
Duration: uptime,
Identities: make([]*dto.Identity, len(t.ids)),
Identities: make([]*dto.Identity, 0),
IpInfo: t.state.IpInfo,
LogLevel: t.state.LogLevel,
ServiceVersion: Version,
@@ -123,8 +124,15 @@ func (t *RuntimeState) ToStatus() dto.TunnelStatus {
i := 0
for _, id := range t.ids {
cid := Clean(id)
clean.Identities[i] = &cid
if onlyInitialized {
if id.CId != nil && id.CId.Loaded {
cid := Clean(id)
clean.Identities = append(clean.Identities, &cid)
}
} else {
cid := Clean(id)
clean.Identities = append(clean.Identities, &cid)
}
i++
}
@@ -231,31 +239,35 @@ func (t *RuntimeState) LoadIdentity(id *Id) {
}
log.Infof("loading identity %s[%s]", id.Name, id.FingerPrint)
now := time.Now()
id.CId = cziti.LoadZiti(id.Path(), id.Active)
/*if id.CId == nil {
log.Warnf("connecting to identity with fingerprint [%s] did not error but no context was returned", id.FingerPrint)
return
}*/
log.Infof("loading identity for %s[%s] completed taking %f seconds", id.Name, id.FingerPrint, time.Now().Sub(now).Seconds())
go func(){
statusChange := <- id.CId.StatusChanges //wait for the response
log.Infof("status change! %d", statusChange)
id.ControllerVersion = id.CId.Version
id.CId.Fingerprint = id.FingerPrint
id.CId.Loaded = true
id.Config.ZtAPI = id.CId.Controller()
id.ControllerVersion = id.CId.Version
id.CId.Fingerprint = id.FingerPrint
id.CId.Loaded = true
id.Config.ZtAPI = id.CId.Controller()
// hack for now - if the identity name is '<unknown>' don't set it... :(
if id.CId.Name == "<unknown>" {
log.Debugf("name is set to <unknown> which probably indicates the controller is down - not changing the name")
} else if id.Name != id.CId.Name {
log.Debugf("name changed from %s to %s", id.Name, id.CId.Name)
id.Name = id.CId.Name
}
log.Infof("successfully loaded %s@%s", id.CId.Name, id.CId.Controller())
_, found := t.ids[id.FingerPrint]
if !found {
t.ids[id.FingerPrint] = id //add this identity to the list of known ids
}
// hack for now - if the identity name is '<unknown>' don't set it... :(
if id.CId.Name == "<unknown>" {
log.Debugf("name is set to <unknown> which probably indicates the controller is down - not changing the name")
} else if id.Name != id.CId.Name {
log.Debugf("name changed from %s to %s", id.Name, id.CId.Name)
id.Name = id.CId.Name
}
log.Infof("successfully loaded %s@%s", id.CId.Name, id.CId.Controller())
_, found := t.ids[id.FingerPrint]
if !found {
t.ids[id.FingerPrint] = id //add this identity to the list of known ids
}
events.broadcast <- dto.IdentityEvent{
ActionEvent: IDENTITY_ADDED,
Id: id.Identity,
}
log.Infof("connecting identity completed: %s[%s]", id.Name, id.FingerPrint)
}()
}
func (t *RuntimeState) LoadConfig() {