egommerce/go-api-pkg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added Consul KV base support

Browse Source
This commit is contained in:
Piotr Biernat 2022-12-05 19:23:32 +01:00
parent faace02cb4
commit 39a8327463
1 changed files with 28 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
consul/discovery.go
View File

@ -10,31 +10,33 @@ import (
)
type Service struct {
AppID string
Name string
Domain string
Address string
Port int
TTL time.Duration
ConsulAgent *consul.Agent
Name string
Address string
appID string
domain string
port int
ttl time.Duration
agent *consul.Agent
kv *consul.KV
}
var ErrServiceUnavailable = fmt.Errorf("Service is unavailable")
func NewService(servAddr, id, name, hostname, domain string, appPort int) (*Service, error) {
s := new(Service)
s.AppID = id
s.Name = name
s.Address = hostname
s.Domain = domain
s.Port = appPort
s.TTL = time.Second * 15
s.appID = id
s.domain = domain
s.port = appPort
s.ttl = time.Second * 15
client, err := consul.NewClient(newClientConfig(servAddr))
if err != nil {
return nil, err
}
s.ConsulAgent = client.Agent()
s.agent = client.Agent()
s.kv = client.KV()
return s, nil
}
@ -47,11 +49,11 @@ func newClientConfig(serverAddr string) *consul.Config {
}
func (s *Service) GetID() string {
return fmt.Sprintf("%s_%s", s.Name, s.AppID)
return fmt.Sprintf("%s_%s", s.Name, s.appID)
}
func (s *Service) GetFullAddr() string {
return fmt.Sprintf("http://%s:%d/", s.Address, s.Port)
return fmt.Sprintf("http://%s:%d/", s.Address, s.port)
}
func (s *Service) Register() error {
@ -59,14 +61,14 @@ func (s *Service) Register() error {
ID: s.GetID(),
Name: s.Name,
Address: s.Address,
Port: s.Port,
Port: s.port,
Tags: s.getTags(),
Check: &consul.AgentServiceCheck{
TTL: s.TTL.String(),
TTL: s.ttl.String(),
},
}
if err := s.ConsulAgent.ServiceRegister(def); err != nil {
if err := s.agent.ServiceRegister(def); err != nil {
return err
}
@ -81,7 +83,7 @@ func (s *Service) Register() error {
}()
go func() { // TTL
interval := s.TTL - time.Second*2
interval := s.ttl - time.Second*2
ticker := time.NewTicker(interval)
for range ticker.C {
_, err := s.healthCheck()
@ -94,7 +96,11 @@ func (s *Service) Register() error {
return nil
}
func (s *Service) Unregister() error {
return s.ConsulAgent.ServiceDeregister(s.GetID())
return s.agent.ServiceDeregister(s.GetID())
}
func (s *Service) KV() *consul.KV {
return s.kv
}
func (s *Service) healthCheck() (bool, error) {
@ -117,13 +123,13 @@ func (s *Service) healthCheck() (bool, error) {
}()
if alive {
if err := s.ConsulAgent.PassTTL("service:"+s.GetID(), "OK"); err != nil {
if err := s.agent.PassTTL("service:"+s.GetID(), "OK"); err != nil {
return false, err
}
return true, nil
}
if err := s.ConsulAgent.FailTTL("service:"+s.GetID(), ErrServiceUnavailable.Error()); err != nil {
if err := s.agent.FailTTL("service:"+s.GetID(), ErrServiceUnavailable.Error()); err != nil {
return false, err
}
return false, ErrServiceUnavailable
@ -138,7 +144,7 @@ func (s *Service) getTags() []string {
"traefik.http.routers." + s.Name + ".service=" + s.Name,
"traefik.http.routers." + s.Name + ".middlewares=compress,requestid",
"traefik.http.services." + s.Name + ".loadbalancer.server.scheme=http",
"traefik.http.services." + s.Name + ".loadbalancer.server.port=" + strconv.Itoa(s.Port),
"traefik.http.services." + s.Name + ".loadbalancer.server.port=" + strconv.Itoa(s.port),
"traefik.http.middlewares.compress.compress=true",
"traefik.http.middlewares.requestid.plugin.requestid.headerName=X-Request-ID",
"traefik.tls.certificates.certfile=/certs/client.cert",