always pass a cluster name as a logger field
This commit is contained in:
parent
e26db66cb5
commit
5470f20be4
|
|
@ -79,8 +79,6 @@ type compareStatefulsetResult struct {
|
||||||
|
|
||||||
// New creates a new cluster. This function should be called from a controller.
|
// New creates a new cluster. This function should be called from a controller.
|
||||||
func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec spec.Postgresql, logger *logrus.Entry) *Cluster {
|
func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec spec.Postgresql, logger *logrus.Entry) *Cluster {
|
||||||
lg := logger.WithField("pkg", "cluster").WithField("cluster-name", pgSpec.Name)
|
|
||||||
k8sResources := kubeResources{Secrets: make(map[types.UID]*v1.Secret), Service: make(map[postgresRole]*v1.Service)}
|
|
||||||
orphanDependents := true
|
orphanDependents := true
|
||||||
|
|
||||||
podEventsQueue := cache.NewFIFO(func(obj interface{}) (string, error) {
|
podEventsQueue := cache.NewFIFO(func(obj interface{}) (string, error) {
|
||||||
|
|
@ -95,18 +93,18 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec spec.Postgresql
|
||||||
cluster := &Cluster{
|
cluster := &Cluster{
|
||||||
Config: cfg,
|
Config: cfg,
|
||||||
Postgresql: pgSpec,
|
Postgresql: pgSpec,
|
||||||
logger: lg,
|
|
||||||
pgUsers: make(map[string]spec.PgUser),
|
pgUsers: make(map[string]spec.PgUser),
|
||||||
systemUsers: make(map[string]spec.PgUser),
|
systemUsers: make(map[string]spec.PgUser),
|
||||||
podSubscribers: make(map[spec.NamespacedName]chan spec.PodEvent),
|
podSubscribers: make(map[spec.NamespacedName]chan spec.PodEvent),
|
||||||
kubeResources: k8sResources,
|
kubeResources: kubeResources{Secrets: make(map[types.UID]*v1.Secret), Service: make(map[postgresRole]*v1.Service)},
|
||||||
masterLess: false,
|
masterLess: false,
|
||||||
userSyncStrategy: users.DefaultUserSyncStrategy{},
|
userSyncStrategy: users.DefaultUserSyncStrategy{},
|
||||||
deleteOptions: &metav1.DeleteOptions{OrphanDependents: &orphanDependents},
|
deleteOptions: &metav1.DeleteOptions{OrphanDependents: &orphanDependents},
|
||||||
podEventsQueue: podEventsQueue,
|
podEventsQueue: podEventsQueue,
|
||||||
KubeClient: kubeClient,
|
KubeClient: kubeClient,
|
||||||
teamsAPIClient: teams.NewTeamsAPI(cfg.OpConfig.TeamsAPIUrl, logger.Logger),
|
teamsAPIClient: teams.NewTeamsAPI(cfg.OpConfig.TeamsAPIUrl, logger),
|
||||||
}
|
}
|
||||||
|
cluster.logger = logger.WithField("pkg", "cluster").WithField("cluster-name", cluster.clusterName())
|
||||||
|
|
||||||
return cluster
|
return cluster
|
||||||
}
|
}
|
||||||
|
|
@ -139,7 +137,7 @@ func (c *Cluster) setStatus(status spec.PostgresStatus) {
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.logger.Warningf("could not set status for cluster %q: %v", c.clusterName(), err)
|
c.logger.Warningf("could not set status for the cluster: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -218,7 +216,7 @@ func (c *Cluster) Create() error {
|
||||||
}
|
}
|
||||||
c.logger.Infof("statefulset %q has been successfully created", util.NameFromMeta(ss.ObjectMeta))
|
c.logger.Infof("statefulset %q has been successfully created", util.NameFromMeta(ss.ObjectMeta))
|
||||||
|
|
||||||
c.logger.Info("waiting for cluster being ready")
|
c.logger.Info("waiting for the cluster being ready")
|
||||||
|
|
||||||
if err = c.waitStatefulsetPodsReady(); err != nil {
|
if err = c.waitStatefulsetPodsReady(); err != nil {
|
||||||
c.logger.Errorf("failed to create cluster: %v", err)
|
c.logger.Errorf("failed to create cluster: %v", err)
|
||||||
|
|
@ -575,7 +573,7 @@ func (c *Cluster) processPodEventQueue(stopCh <-chan struct{}) {
|
||||||
return
|
return
|
||||||
default:
|
default:
|
||||||
if _, err := c.podEventsQueue.Pop(cache.PopProcessFunc(c.processPodEvent)); err != nil {
|
if _, err := c.podEventsQueue.Pop(cache.PopProcessFunc(c.processPodEvent)); err != nil {
|
||||||
c.logger.Errorf("error when processing pod event queeue %v", err)
|
c.logger.Errorf("error when processing pod event queue %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -145,7 +145,7 @@ PATRONI_INITDB_PARAMS:
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
default:
|
default:
|
||||||
c.logger.Warnf("unsupported type for initdb configuration item %s: %T", defaultParam)
|
c.logger.Warnf("unsupported type for initdb configuration item %s: %T", defaultParam, defaultParam)
|
||||||
continue PATRONI_INITDB_PARAMS
|
continue PATRONI_INITDB_PARAMS
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -45,7 +45,10 @@ func (c *Controller) dispatchPodEvent(clusterName spec.NamespacedName, event spe
|
||||||
cluster, ok := c.clusters[clusterName]
|
cluster, ok := c.clusters[clusterName]
|
||||||
c.clustersMu.RUnlock()
|
c.clustersMu.RUnlock()
|
||||||
if ok {
|
if ok {
|
||||||
c.logger.Debugf("sending %q event of pod %q to the %q cluster channel", event.EventType, event.PodName, clusterName)
|
c.logger.WithField("cluster-name", clusterName).
|
||||||
|
Debugf("sending %q event of pod %q to the cluster channel",
|
||||||
|
event.EventType,
|
||||||
|
event.PodName)
|
||||||
cluster.ReceivePodEvent(event)
|
cluster.ReceivePodEvent(event)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -131,13 +131,14 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("could not cast to ClusterEvent")
|
return fmt.Errorf("could not cast to ClusterEvent")
|
||||||
}
|
}
|
||||||
logger := c.logger.WithField("worker", event.WorkerID)
|
lg := c.logger.WithField("worker", event.WorkerID)
|
||||||
|
|
||||||
if event.EventType == spec.EventAdd || event.EventType == spec.EventSync {
|
if event.EventType == spec.EventAdd || event.EventType == spec.EventSync {
|
||||||
clusterName = util.NameFromMeta(event.NewSpec.ObjectMeta)
|
clusterName = util.NameFromMeta(event.NewSpec.ObjectMeta)
|
||||||
} else {
|
} else {
|
||||||
clusterName = util.NameFromMeta(event.OldSpec.ObjectMeta)
|
clusterName = util.NameFromMeta(event.OldSpec.ObjectMeta)
|
||||||
}
|
}
|
||||||
|
lg = lg.WithField("cluster-name", clusterName)
|
||||||
|
|
||||||
c.clustersMu.RLock()
|
c.clustersMu.RLock()
|
||||||
cl, clusterFound := c.clusters[clusterName]
|
cl, clusterFound := c.clusters[clusterName]
|
||||||
|
|
@ -146,14 +147,14 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||||
switch event.EventType {
|
switch event.EventType {
|
||||||
case spec.EventAdd:
|
case spec.EventAdd:
|
||||||
if clusterFound {
|
if clusterFound {
|
||||||
logger.Debugf("cluster %q already exists", clusterName)
|
lg.Debugf("cluster already exists")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Infof("creation of the %q cluster started", clusterName)
|
lg.Infof("creation of the cluster started")
|
||||||
|
|
||||||
stopCh := make(chan struct{})
|
stopCh := make(chan struct{})
|
||||||
cl = cluster.New(c.makeClusterConfig(), c.KubeClient, *event.NewSpec, logger)
|
cl = cluster.New(c.makeClusterConfig(), c.KubeClient, *event.NewSpec, lg)
|
||||||
cl.Run(stopCh)
|
cl.Run(stopCh)
|
||||||
|
|
||||||
c.clustersMu.Lock()
|
c.clustersMu.Lock()
|
||||||
|
|
@ -163,36 +164,36 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||||
|
|
||||||
if err := cl.Create(); err != nil {
|
if err := cl.Create(); err != nil {
|
||||||
cl.Error = fmt.Errorf("could not create cluster: %v", err)
|
cl.Error = fmt.Errorf("could not create cluster: %v", err)
|
||||||
logger.Errorf("%v", cl.Error)
|
lg.Error(cl.Error)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Infof("cluster %q has been created", clusterName)
|
lg.Infoln("cluster has been created")
|
||||||
case spec.EventUpdate:
|
case spec.EventUpdate:
|
||||||
logger.Infof("update of the %q cluster started", clusterName)
|
lg.Infoln("update of the cluster started")
|
||||||
|
|
||||||
if !clusterFound {
|
if !clusterFound {
|
||||||
logger.Warnf("cluster %q does not exist", clusterName)
|
lg.Warnln("cluster does not exist")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
if err := cl.Update(event.NewSpec); err != nil {
|
if err := cl.Update(event.NewSpec); err != nil {
|
||||||
cl.Error = fmt.Errorf("could not update cluster: %v", err)
|
cl.Error = fmt.Errorf("could not update cluster: %v", err)
|
||||||
logger.Errorf("%v", cl.Error)
|
lg.Error(cl.Error)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
cl.Error = nil
|
cl.Error = nil
|
||||||
logger.Infof("cluster %q has been updated", clusterName)
|
lg.Infoln("cluster has been updated")
|
||||||
case spec.EventDelete:
|
case spec.EventDelete:
|
||||||
logger.Infof("deletion of the %q cluster started", clusterName)
|
lg.Infof("deletion of the %q cluster started", clusterName)
|
||||||
if !clusterFound {
|
if !clusterFound {
|
||||||
logger.Errorf("unknown cluster: %q", clusterName)
|
lg.Errorf("unknown cluster: %q", clusterName)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := cl.Delete(); err != nil {
|
if err := cl.Delete(); err != nil {
|
||||||
logger.Errorf("could not delete cluster %q: %v", clusterName, err)
|
lg.Errorf("could not delete cluster: %v", err)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
close(c.stopChs[clusterName])
|
close(c.stopChs[clusterName])
|
||||||
|
|
@ -202,14 +203,14 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||||
delete(c.stopChs, clusterName)
|
delete(c.stopChs, clusterName)
|
||||||
c.clustersMu.Unlock()
|
c.clustersMu.Unlock()
|
||||||
|
|
||||||
logger.Infof("cluster %q has been deleted", clusterName)
|
lg.Infof("cluster has been deleted")
|
||||||
case spec.EventSync:
|
case spec.EventSync:
|
||||||
logger.Infof("syncing of the %q cluster started", clusterName)
|
lg.Infof("syncing of the cluster started")
|
||||||
|
|
||||||
// no race condition because a cluster is always processed by single worker
|
// no race condition because a cluster is always processed by single worker
|
||||||
if !clusterFound {
|
if !clusterFound {
|
||||||
stopCh := make(chan struct{})
|
stopCh := make(chan struct{})
|
||||||
cl = cluster.New(c.makeClusterConfig(), c.KubeClient, *event.NewSpec, logger)
|
cl = cluster.New(c.makeClusterConfig(), c.KubeClient, *event.NewSpec, lg)
|
||||||
cl.Run(stopCh)
|
cl.Run(stopCh)
|
||||||
|
|
||||||
c.clustersMu.Lock()
|
c.clustersMu.Lock()
|
||||||
|
|
@ -220,12 +221,12 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||||
|
|
||||||
if err := cl.Sync(); err != nil {
|
if err := cl.Sync(); err != nil {
|
||||||
cl.Error = fmt.Errorf("could not sync cluster %q: %v", clusterName, err)
|
cl.Error = fmt.Errorf("could not sync cluster %q: %v", clusterName, err)
|
||||||
logger.Errorf("%v", cl.Error)
|
lg.Error(cl.Error)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
cl.Error = nil
|
cl.Error = nil
|
||||||
|
|
||||||
logger.Infof("cluster %q has been synced", clusterName)
|
lg.Infof("cluster has been synced")
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|
@ -273,7 +274,9 @@ func (c *Controller) queueClusterEvent(old, new *spec.Postgresql, eventType spec
|
||||||
}
|
}
|
||||||
|
|
||||||
if clusterError != nil && eventType != spec.EventDelete {
|
if clusterError != nil && eventType != spec.EventDelete {
|
||||||
c.logger.Debugf("skipping %q event for invalid cluster %q (reason: %v)", eventType, clusterName, clusterError)
|
c.logger.
|
||||||
|
WithField("cluster-name", clusterName).
|
||||||
|
Debugf("skipping %q event for the invalid cluster: %v", eventType, clusterError)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -287,10 +290,11 @@ func (c *Controller) queueClusterEvent(old, new *spec.Postgresql, eventType spec
|
||||||
}
|
}
|
||||||
//TODO: if we delete cluster, discard all the previous events for the cluster
|
//TODO: if we delete cluster, discard all the previous events for the cluster
|
||||||
|
|
||||||
|
lg := c.logger.WithField("worker", workerID).WithField("cluster-name", clusterName)
|
||||||
if err := c.clusterEventQueues[workerID].Add(clusterEvent); err != nil {
|
if err := c.clusterEventQueues[workerID].Add(clusterEvent); err != nil {
|
||||||
c.logger.WithField("worker", workerID).Errorf("error when queueing cluster event: %v", clusterEvent)
|
lg.Errorf("error when queueing cluster event: %v", clusterEvent)
|
||||||
}
|
}
|
||||||
c.logger.WithField("worker", workerID).Infof("%q of the %q cluster has been queued", eventType, clusterName)
|
lg.Infof("%q event has been queued", eventType)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Controller) postgresqlAdd(obj interface{}) {
|
func (c *Controller) postgresqlAdd(obj interface{}) {
|
||||||
|
|
|
||||||
|
|
@ -51,7 +51,7 @@ type API struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewTeamsAPI creates an object to query the team API.
|
// NewTeamsAPI creates an object to query the team API.
|
||||||
func NewTeamsAPI(url string, log *logrus.Logger) *API {
|
func NewTeamsAPI(url string, log *logrus.Entry) *API {
|
||||||
t := API{
|
t := API{
|
||||||
url: strings.TrimRight(url, "/"),
|
url: strings.TrimRight(url, "/"),
|
||||||
httpClient: &http.Client{},
|
httpClient: &http.Client{},
|
||||||
|
|
|
||||||
|
|
@ -11,7 +11,7 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
logger = logrus.New()
|
logger = logrus.New().WithField("pkg", "teamsapi")
|
||||||
token = "ec45b1cfbe7100c6315d183a3eb6cec0M2U1LWJkMzEtZDgzNzNmZGQyNGM3IiwiYXV0aF90aW1lIjoxNDkzNzMwNzQ1LCJpc3MiOiJodHRwcz"
|
token = "ec45b1cfbe7100c6315d183a3eb6cec0M2U1LWJkMzEtZDgzNzNmZGQyNGM3IiwiYXV0aF90aW1lIjoxNDkzNzMwNzQ1LCJpc3MiOiJodHRwcz"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue