overflow_cloud/probe
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-04-13 20:59:46 +09:00
parent 0fc01915ae
commit 45dc0f9920
12 changed files with 385 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
auth.json Normal file
View File

@ -0,0 +1,3 @@
{
"tempKey": "6a9937503f1111e8b0460242ac120002"
}

23
auth/authenticator.go
View File

@ -53,8 +53,8 @@ func (a *Authenticator) EndableStart() (<-chan error, error) {
}
}
if nil != a.authConfig.DenyDate {
return nil, fmt.Errorf("cannot start because this probe have been denied from overFlow at %s", a.authConfig.DenyDate.String())
if nil != a.authConfig.DeniedDate {
return nil, fmt.Errorf("cannot start because this probe have been denied from overFlow at %s", a.authConfig.DeniedDate.String())
}
a.accpetedChan = make(chan string, 1)
@ -95,7 +95,7 @@ func (a *Authenticator) logHeader() string {
}
func (a *Authenticator) initService() (crr.RPCInvoker, error) {
cdr.RegisterResource("AccpetedChan", a.accpetedChan)
cdr.RegisterResource("AcceptedChan", a.accpetedChan)
cdr.RegisterResource("DeniedChan", a.deniedChan)
services, err := cdr.GetInstancesByAnnotationType(occa.RPCServiceAnnotationType)
@ -128,9 +128,12 @@ func (a *Authenticator) initClient() error {
header[ocncc.HTTPRequestHeaderKey_NoAuthProbe_Info] = []string{rh}
}
centralURL := fmt.Sprintf("ws://%s:%d", a.Config.Central.Host, a.Config.Central.Port)
centralURL := fmt.Sprintf("ws://%s:%d%s", a.Config.Central.Host, a.Config.Central.Port, ocncc.HTTPEntry_Auth)
connector.URL = path.Join(centralURL, ocncc.HTTPEntry_Auth)
logging.Logger().Debugf("%s central %s", a.logHeader(), centralURL)
connector.Name = "Authenticator"
connector.URL = centralURL
connector.RequestHeader = header
connector.ResponseHandler = func(res *http.Response) {
switch a.authConfig.State() {
@ -175,6 +178,14 @@ func (a *Authenticator) handleAuthenticator(endChan chan<- error) {
return
}
logging.Logger().Infof("%s accepted by central", a.logHeader())
n := time.Now()
a.authConfig.AcceptedDate = &n
err = configuration.Save(a.authConfig, path.Join(a.ConfigDir, ocnc.ConfigFileName), true)
if nil != err {
logging.Logger().Errorf("%s %v", a.logHeader(), err)
}
a.Config.Probe.Key = &probeKey
err = configuration.Save(a.Config, path.Join(a.ConfigDir, ocpc.ConfigFileName), true)
if nil != err {
@ -187,7 +198,7 @@ func (a *Authenticator) handleAuthenticator(endChan chan<- error) {
}
logging.Logger().Infof("%s denied by central", a.logHeader())
n := time.Now()
a.authConfig.DenyDate = &n
a.authConfig.DeniedDate = &n
err = configuration.Save(a.authConfig, path.Join(a.ConfigDir, ocnc.ConfigFileName), true)
if nil != err {
logging.Logger().Errorf("%s %v", a.logHeader(), err)

32
auth/service/NoAuthProbeService.go Normal file
View File

@ -0,0 +1,32 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var NoAuthProbeServiceType = reflect.TypeOf((*NoAuthProbeService)(nil))
func init() {
cdr.RegisterType(NoAuthProbeServiceType)
}
type NoAuthProbeService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
AcceptedChan chan<- string `annotation:"@Resource(name='AcceptedChan')"`
DeniedChan chan<- struct{} `annotation:"@Resource(name='DeniedChan')"`
}
func (s *NoAuthProbeService) Accept(probeKey string) error {
s.AcceptedChan <- probeKey
return nil
}
func (s *NoAuthProbeService) Deny() error {
s.DeniedChan <- struct{}{}
return nil
}

32
auth/service/auth-service.go
View File

@ -1,32 +0,0 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var AuthServiceType = reflect.TypeOf((*AuthService)(nil))
func init() {
cdr.RegisterType(AuthServiceType)
}
type AuthService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
AcceptedChan chan<- string
DeniedChan chan<- struct{}
}
func (as *AuthService) Accept(probeKey string) error {
as.AcceptedChan <- probeKey
return nil
}
func (as *AuthService) Deny() error {
as.DeniedChan <- struct{}{}
return nil
}

13
config.json
View File

@ -5,8 +5,13 @@
},
"central": {
"host": "127.0.0.1",
"port": 19190,
"port": 19091,
"connector": {
"name": "",
"network": "",
"address": "",
"concurrency": 0,
"keepAlive": 0,
"handshakeTimeout": 0,
"reconnectInterval": 5,
"reconnectTryTime": 10,
@ -18,7 +23,9 @@
"pongTimeout": 60,
"pingTimeout": 10,
"pingPeriod": 9,
"enableCompression": false
"enableCompression": false,
"url": "",
"subprotocols": null
},
"proxy": {
"host": "",
@ -29,7 +36,7 @@
}
},
"probe": {
"key": "866f9be0333311e8b7230242ac120004"
"key": "7691f1a13f1111e8b0460242ac120002"
},
"paths": {
"root": "/project/overFlow/probe"

1
main.go
View File

@ -53,6 +53,7 @@ func main() {
if nil != err {
logging.Logger().Panic(err)
}
}
// err := s.ListenAndServe()

187
probe/probe.go Normal file
View File

@ -0,0 +1,187 @@
package probe
import (
"context"
"fmt"
"net/http"
"path"
"sync"
"time"
"git.loafle.net/commons/configuration-go"
cdr "git.loafle.net/commons/di-go/registry"
logging "git.loafle.net/commons/logging-go"
crc "git.loafle.net/commons/rpc-go/client"
crpj "git.loafle.net/commons/rpc-go/protocol/json"
crr "git.loafle.net/commons/rpc-go/registry"
csswc "git.loafle.net/commons/server-go/socket/web/client"
occa "git.loafle.net/overflow/commons-go/core/annotation"
ocnc "git.loafle.net/overflow/commons-go/noauthprobe/config"
ocncc "git.loafle.net/overflow/commons-go/noauthprobe/constants"
ocpc "git.loafle.net/overflow/commons-go/probe/config"
"git.loafle.net/overflow/probe/auth/info"
_ "git.loafle.net/overflow/probe/auth/service"
"git.loafle.net/overflow/probe/config"
)
type Probe struct {
Config *config.Config
ConfigDir string
client *crc.Client
stopChan chan struct{}
stopWg sync.WaitGroup
}
func (p *Probe) Start() error {
if p.stopChan != nil {
return fmt.Errorf("already running. Stop it before starting it again")
}
if err := p.initClient(); nil != err {
return err
}
if err := p.client.Start(); nil != err {
return err
}
p.stopChan = make(chan struct{})
p.stopWg.Add(1)
go p.handleProbe()
return nil
}
func (p *Probe) Stop(ctx context.Context) error {
if p.stopChan == nil {
return fmt.Errorf("must be started before stopping it")
}
close(p.stopChan)
p.stopWg.Wait()
p.stopChan = nil
return nil
}
func (p *Probe) logHeader() string {
return "Probe:"
}
func (p *Probe) initService() (crr.RPCInvoker, error) {
services, err := cdr.GetInstancesByAnnotationType(occa.RPCServiceAnnotationType)
if nil != err {
return nil, err
}
rpcRegistry := crr.NewRPCRegistry()
rpcRegistry.RegisterServices(services...)
return rpcRegistry, nil
}
func (p *Probe) initClient() error {
_connector := p.Config.Central.Connector.Clone()
connector := _connector.(*csswc.Connectors)
header := make(map[string][]string)
switch p.authConfig.State() {
case ocnc.AuthStateTypeRegisterd:
header[ocncc.HTTPRequestHeaderKey_NoAuthProbe_Method] = []string{ocncc.HTTPRequestHeaderValue_NoAuthProbe_Method_Connect}
header[ocncc.HTTPRequestHeaderKey_NoAuthProbe_TempProbeKey] = []string{*p.authConfig.TempKey}
default:
rh, err := info.GetRegistHeader(p.Config.Account.APIKey)
if nil != err {
return err
}
header[ocncc.HTTPRequestHeaderKey_NoAuthProbe_Method] = []string{ocncc.HTTPRequestHeaderValue_NoAuthProbe_Method_Regist}
header[ocncc.HTTPRequestHeaderKey_NoAuthProbe_Info] = []string{rh}
}
centralURL := fmt.Sprintf("ws://%s:%d%s", p.Config.Central.Host, p.Config.Central.Port, ocncc.HTTPEntry_Auth)
logging.Logger().Debugf("%s central %s", p.logHeader(), centralURL)
connector.URL = centralURL
connector.RequestHeader = header
connector.ResponseHandler = func(res *http.Response) {
switch p.authConfig.State() {
case ocnc.AuthStateTypeNotRegisterd:
tempProbeKey := res.Header.Get(ocncc.HTTPResponseHeaderKey_NoAuthProbe_SetTempProbeKey)
p.tempKeyChan <- tempProbeKey
default:
}
}
rpcInvoker, err := p.initService()
if nil != err {
return err
}
codec := crpj.NewClientCodec()
p.client = &crc.Client{
Connector: connector,
Codec: codec,
RPCInvoker: rpcInvoker,
Name: "Authenticator",
}
return nil
}
func (p *Probe) handleProbe(endChan chan<- error) {
var err error
defer func() {
if nil != p.client {
err = p.client.Stop(context.Background())
}
p.stopWg.Done()
endChan <- err
}()
for {
select {
case probeKey, ok := <-p.accpetedChan:
if !ok {
return
}
logging.Logger().Infof("%s accepted by central", p.logHeader())
p.Config.Probe.Key = &probeKey
err = configuration.Save(p.Config, path.Join(p.ConfigDir, ocpc.ConfigFileName), true)
if nil != err {
logging.Logger().Errorf("%s %v", p.logHeader(), err)
}
return
case _, ok := <-p.deniedChan:
if !ok {
return
}
logging.Logger().Infof("%s denied by central", p.logHeader())
n := time.Now()
p.authConfig.DenyDate = &n
err = configuration.Save(p.authConfig, path.Join(p.ConfigDir, ocnc.ConfigFileName), true)
if nil != err {
logging.Logger().Errorf("%s %v", p.logHeader(), err)
}
return
case tempKey, ok := <-p.tempKeyChan:
if !ok {
return
}
logging.Logger().Infof("%s registered by central", p.logHeader())
p.authConfig.TempKey = &tempKey
err = configuration.Save(p.authConfig, path.Join(p.ConfigDir, ocnc.ConfigFileName), true)
if nil != err {
logging.Logger().Errorf("%s %v", p.logHeader(), err)
return
}
case <-p.stopChan:
return
}
}
}

24
service/ContainerService.go Normal file
View File

@ -0,0 +1,24 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var ContainerServiceType = reflect.TypeOf((*ContainerService)(nil))
func init() {
cdr.RegisterType(ContainerServiceType)
}
type ContainerService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
}
func (cs *ContainerService) Accept() error {
return nil
}

24
service/CrawlerService.go Normal file
View File

@ -0,0 +1,24 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var CrawlerServiceType = reflect.TypeOf((*CrawlerService)(nil))
func init() {
cdr.RegisterType(CrawlerServiceType)
}
type CrawlerService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
}
func (cs *CrawlerService) Accept() error {
return nil
}

24
service/DiscoveryService.go Normal file
View File

@ -0,0 +1,24 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var DiscoveryServiceType = reflect.TypeOf((*DiscoveryService)(nil))
func init() {
cdr.RegisterType(DiscoveryServiceType)
}
type DiscoveryService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
}
func (cs *DiscoveryService) Accept() error {
return nil
}

24
service/ProbeService.go Normal file
View File

@ -0,0 +1,24 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var ProbeServiceType = reflect.TypeOf((*ProbeService)(nil))
func init() {
cdr.RegisterType(ProbeServiceType)
}
type ProbeService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
}
func (cs *ProbeService) Accept() error {
return nil
}

13
service/service.go Normal file
View File

@ -0,0 +1,13 @@
package service
func InitPackage() {
}
func StartPackage() {
}
func StopPackage() {
}
func DestroyPackage() {
}