wip

armadaproject · d80tb7 · Jan 20, 2025 · Jan 2, 2025 · Jan 3, 2025 · Jan 3, 2025
commit da74217b987efc8e8aaf44dc1ceda68afa29defb
diff --git a/internal/common/etcdhealth/etcdhealth_test.go b/internal/common/etcdhealth/etcdhealth_test.go
@@ -26,7 +26,7 @@ func TestEtcdReplicaHealthMonitor(t *testing.T) {
 	ctx, cancel := armadacontext.WithCancel(armadacontext.Background())
 	defer cancel()
 	g, ctx := armadacontext.ErrGroup(ctx)
-	g.Go(func() error { return hm.Run(ctx, logging.NewLogger()) })
+	g.Go(func() error { return hm.Run(ctx, logging.StdLogger()) })
 
 	// Should still be unavailable due to missing metrics.
 	hm.BlockUntilNextMetricsCollection(ctx)

diff --git a/internal/common/ingest/controlplaneevents/utils.go b/internal/common/ingest/controlplaneevents/utils.go
@@ -7,7 +7,7 @@ import (
 	commonmetrics "github.com/armadaproject/armada/internal/common/ingest/metrics"
 	"github.com/armadaproject/armada/internal/common/ingest/utils"
 	log "github.com/armadaproject/armada/internal/common/logging"
-
+	"github.com/armadaproject/armada/pkg/controlplaneevents"
 )
 
 func MessageUnmarshaller(msg pulsar.ConsumerMessage, metrics *commonmetrics.Metrics) *utils.EventsWithIds[*controlplaneevents.Event] {

diff --git a/internal/eventingester/store/eventstore.go b/internal/eventingester/store/eventstore.go
@@ -1,6 +1,7 @@
 package store
 
 import (
+	"github.com/armadaproject/armada/internal/eventingester/model"
 	"regexp"
 	"time"
 
@@ -10,9 +11,7 @@ import (
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 	"github.com/armadaproject/armada/internal/common/ingest"
 	log "github.com/armadaproject/armada/internal/common/logging"
-	"github
 	"github.com/armadaproject/armada/internal/eventingester/configuration"
-	log "github.com/armadaproject/armada/internal/common/logging"
 )
 
 const (

diff --git a/internal/executor/service/cluster_allocation.go b/internal/executor/service/cluster_allocation.go
@@ -48,7 +48,7 @@ func (allocationService *ClusterAllocationService) AllocateSpareClusterCapacity(
 	// If a health monitor is provided, avoid leasing jobs when the cluster is unhealthy.
 	if allocationService.clusterHealthMonitor != nil {
 		if ok, reason, err := allocationService.clusterHealthMonitor.IsHealthy(); err != nil {
-			logging.WithStacktrace(logging.NewLogger(), err).Error("failed to check cluster health")
+			logging.WithStacktrace(err).Error("failed to check cluster health")
 			return
 		} else if !ok {
 			log.Warnf("cluster is not healthy; will not request more jobs: %s", reason)

diff --git a/internal/lookoutv2/application.go b/internal/lookoutv2/application.go
@@ -47,7 +47,7 @@ func Serve(configuration configuration.LookoutV2Config) error {
 	// create new service API
 	api := operations.NewLookoutAPI(swaggerSpec)
 
-	logger := logging.NewLogger()
+	logger := logging.StdLogger()
 
 	api.Logger = logger.Debugf
 

diff --git a/internal/scheduler/api.go b/internal/scheduler/api.go
@@ -16,7 +16,6 @@ import (
 	"github.com/armadaproject/armada/internal/common/armadaerrors"
 	"github.com/armadaproject/armada/internal/common/auth"
 	"github.com/armadaproject/armada/internal/common/compress"
-	"github.com/armadaproject/armada/internal/common/logging"
 	log "github.com/armadaproject/armada/internal/common/logging"
 	"github.com/armadaproject/armada/internal/common/maps"
 	"github.com/armadaproject/armada/internal/common/pulsarutils"
@@ -371,7 +370,7 @@ func (srv *ExecutorApi) executorFromLeaseRequest(ctx *armadacontext.Context, req
 	now := srv.clock.Now().UTC()
 	for _, nodeInfo := range req.Nodes {
 		if node, err := executorapi.NewNodeFromNodeInfo(nodeInfo, req.ExecutorId, srv.allowedPriorities, now); err != nil {
-			logging.WithStacktrace(ctx, err).Warnf(
+			ctx.Logger().WithStacktrace(err).Warnf(
 				"skipping node %s from executor %s", nodeInfo.GetName(), req.GetExecutorId(),
 			)
 		} else {

diff --git a/internal/scheduler/metrics.go b/internal/scheduler/metrics.go
@@ -10,7 +10,6 @@ import (
 	"k8s.io/utils/clock"
 
 	"github.com/armadaproject/armada/internal/common/armadacontext"
-	"github.com/armadaproject/armada/internal/common/logging"
 	armadamaps "github.com/armadaproject/armada/internal/common/maps"
 	commonmetrics "github.com/armadaproject/armada/internal/common/metrics"
 	"github.com/armadaproject/armada/internal/common/resource"
@@ -103,8 +102,8 @@ func (c *MetricsCollector) Run(ctx *armadacontext.Context) error {
 		case <-ticker.C():
 			err := c.refresh(ctx)
 			if err != nil {
-				logging.
-					WithStacktrace(ctx, err).
+				ctx.Logger().
+					WithStacktrace(err).
 					Warnf("error refreshing metrics state")
 			}
 		}

diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
@@ -13,7 +13,6 @@ import (
 	"k8s.io/utils/clock"
 
 	"github.com/armadaproject/armada/internal/common/armadacontext"
-	"github.com/armadaproject/armada/internal/common/logging"
 	protoutil "github.com/armadaproject/armada/internal/common/proto"
 	"github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
@@ -153,7 +152,7 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 				syncContext, cancel := armadacontext.WithTimeout(ctx, 5*time.Minute)
 				err := s.ensureDbUpToDate(syncContext, 1*time.Second)
 				if err != nil {
-					logging.WithStacktrace(ctx, err).Error("could not become leader")
+					ctx.Logger().WithStacktrace(err).Error("could not become leader")
 					leaderToken = leader.InvalidLeaderToken()
 				} else {
 					fullUpdate = true
@@ -175,7 +174,7 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 
 			result, err := s.cycle(ctx, fullUpdate, leaderToken, shouldSchedule)
 			if err != nil {
-				logging.WithStacktrace(ctx, err).Error("scheduling cycle failure")
+				ctx.Logger().WithStacktrace(err).Error("scheduling cycle failure")
 				leaderToken = leader.InvalidLeaderToken()
 			}
 
@@ -981,7 +980,9 @@ func (s *Scheduler) initialise(ctx *armadacontext.Context) error {
 			return nil
 		default:
 			if _, _, err := s.syncState(ctx, true); err != nil {
-				logging.WithStacktrace(ctx, err).Error("failed to initialise; trying again in 1 second")
+				ctx.Logger().
+					WithStacktrace(err).
+					Error("failed to initialise; trying again in 1 second")
 				time.Sleep(1 * time.Second)
 			} else {
 				ctx.Info("initialisation succeeded")
@@ -1008,7 +1009,9 @@ func (s *Scheduler) ensureDbUpToDate(ctx *armadacontext.Context, pollInterval ti
 		default:
 			numSent, err = s.publisher.PublishMarkers(ctx, groupId)
 			if err != nil {
-				logging.WithStacktrace(ctx, err).Error("Error sending marker messages to pulsar")
+				ctx.Logger().
+					WithStacktrace(err).
+					Error("Error sending marker messages to pulsar")
 				s.clock.Sleep(pollInterval)
 			} else {
 				messagesSent = true
@@ -1024,8 +1027,8 @@ func (s *Scheduler) ensureDbUpToDate(ctx *armadacontext.Context, pollInterval ti
 		default:
 			numReceived, err := s.jobRepository.CountReceivedPartitions(ctx, groupId)
 			if err != nil {
-				logging.
-					WithStacktrace(ctx, err).
+				ctx.Logger().
+					WithStacktrace(err).
 					Error("Error querying the database or marker messages")
 			}
 			if numSent == numReceived {

diff --git a/internal/scheduler/schedulerapp.go b/internal/scheduler/schedulerapp.go
@@ -25,7 +25,6 @@ import (
 	dbcommon "github.com/armadaproject/armada/internal/common/database"
 	grpcCommon "github.com/armadaproject/armada/internal/common/grpc"
 	"github.com/armadaproject/armada/internal/common/health"
-	"github.com/armadaproject/armada/internal/common/logging"
 	log "github.com/armadaproject/armada/internal/common/logging"
 	"github.com/armadaproject/armada/internal/common/profiling"
 	"github.com/armadaproject/armada/internal/common/pulsarutils"
@@ -118,8 +117,8 @@ func Run(config schedulerconfig.Configuration) error {
 	defer func() {
 		err := conn.Close()
 		if err != nil {
-			logging.
-				WithStacktrace(ctx, err).
+			ctx.Logger().
+				WithStacktrace(err).
 				Warnf("Armada api client didn't close down cleanly")
 		}
 	}()

diff --git a/internal/scheduler/scheduling/preempting_queue_scheduler.go b/internal/scheduler/scheduling/preempting_queue_scheduler.go
@@ -99,7 +99,7 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 	scheduledJobsById := make(map[string]*schedulercontext.JobSchedulingContext)
 
 	// Evict preemptible jobs.
-	ctx.WithField("stage", "scheduling-algo").Infof("Evicting preemptible jobs")
+	ctx.Logger().WithField("stage", "scheduling-algo").Infof("Evicting preemptible jobs")
 	evictorResult, inMemoryJobRepo, err := sch.evict(
 		armadacontext.WithLogField(ctx, "stage", "evict for resource balancing"),
 		NewNodeEvictor(
@@ -144,14 +144,14 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 	if err != nil {
 		return nil, err
 	}
-	ctx.WithField("stage", "scheduling-algo").Info("Finished evicting preemptible jobs")
+	ctx.Logger().WithField("stage", "scheduling-algo").Info("Finished evicting preemptible jobs")
 	for _, jctx := range evictorResult.EvictedJctxsByJobId {
 		preemptedJobsById[jctx.JobId] = jctx
 	}
 	maps.Copy(sch.nodeIdByJobId, evictorResult.NodeIdByJobId)
 
 	// Re-schedule evicted jobs/schedule new jobs.
-	ctx.WithField("stage", "scheduling-algo").Info("Performing initial scheduling of jobs onto nodes")
+	ctx.Logger().WithField("stage", "scheduling-algo").Info("Performing initial scheduling of jobs onto nodes")
 	schedulerResult, err := sch.schedule(
 		armadacontext.WithLogField(ctx, "stage", "re-schedule after balancing eviction"),
 		inMemoryJobRepo,
@@ -162,7 +162,7 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 	if err != nil {
 		return nil, err
 	}
-	ctx.WithField("stage", "scheduling-algo").Info("Finished initial scheduling of jobs onto nodes")
+	ctx.Logger().WithField("stage", "scheduling-algo").Info("Finished initial scheduling of jobs onto nodes")
 	for _, jctx := range schedulerResult.ScheduledJobs {
 		if _, ok := preemptedJobsById[jctx.JobId]; ok {
 			delete(preemptedJobsById, jctx.JobId)
@@ -173,7 +173,7 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 	maps.Copy(sch.nodeIdByJobId, schedulerResult.NodeIdByJobId)
 
 	// Evict jobs on oversubscribed nodes.
-	ctx.WithField("stage", "scheduling-algo").Info("Evicting jobs from oversubscribed nodes")
+	ctx.Logger().WithField("stage", "scheduling-algo").Info("Evicting jobs from oversubscribed nodes")
 	reevictResult, inMemoryJobRepo, err := sch.evict(
 		armadacontext.WithLogField(ctx, "stage", "evict oversubscribed"),
 		NewOversubscribedEvictor(
@@ -185,7 +185,7 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 	if err != nil {
 		return nil, err
 	}
-	ctx.WithField("stage", "scheduling-algo").Info("Finished evicting jobs from oversubscribed nodes")
+	ctx.Logger().WithField("stage", "scheduling-algo").Info("Finished evicting jobs from oversubscribed nodes")
 	scheduledAndEvictedJobsById := armadamaps.FilterKeys(
 		scheduledJobsById,
 		func(jobId string) bool {
@@ -205,7 +205,7 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 	// Re-schedule evicted jobs/schedule new jobs.
 	// Only necessary if a non-zero number of jobs were evicted.
 	if len(reevictResult.EvictedJctxsByJobId) > 0 {
-		ctx.WithField("stage", "scheduling-algo").Info("Performing second scheduling ")
+		ctx.Logger().WithField("stage", "scheduling-algo").Info("Performing second scheduling ")
 		rescheduleSchedulerResult, rescheduleErr := sch.schedule(
 			armadacontext.WithLogField(ctx, "stage", "schedule after oversubscribed eviction"),
 			inMemoryJobRepo,
@@ -217,7 +217,7 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 		if rescheduleErr != nil {
 			return nil, rescheduleErr
 		}
-		ctx.WithField("stage", "scheduling-algo").Info("Finished second scheduling pass")
+		ctx.Logger().WithField("stage", "scheduling-algo").Info("Finished second scheduling pass")
 		for _, jctx := range rescheduleSchedulerResult.ScheduledJobs {
 			if _, ok := preemptedJobsById[jctx.JobId]; ok {
 				delete(preemptedJobsById, jctx.JobId)
@@ -231,14 +231,14 @@ func (sch *PreemptingQueueScheduler) Schedule(ctx *armadacontext.Context) (*Sche
 
 	preemptedJobs := maps.Values(preemptedJobsById)
 	scheduledJobs := maps.Values(scheduledJobsById)
-	ctx.WithField("stage", "scheduling-algo").Infof("Unbinding %d preempted and %d evicted jobs", len(preemptedJobs), len(maps.Values(scheduledAndEvictedJobsById)))
+	ctx.Logger().WithField("stage", "scheduling-algo").Infof("Unbinding %d preempted and %d evicted jobs", len(preemptedJobs), len(maps.Values(scheduledAndEvictedJobsById)))
 	if err := sch.unbindJobs(append(
 		slices.Clone(preemptedJobs),
 		maps.Values(scheduledAndEvictedJobsById)...),
 	); err != nil {
 		return nil, err
 	}
-	ctx.WithField("stage", "scheduling-algo").Infof("Finished unbinding preempted and evicted jobs")
+	ctx.Logger().WithField("stage", "scheduling-algo").Infof("Finished unbinding preempted and evicted jobs")
 
 	PopulatePreemptionDescriptions(preemptedJobs, scheduledJobs)
 	schedulercontext.PrintJobSummary(ctx, "Preempting running jobs;", preemptedJobs)

diff --git a/internal/scheduler/submitcheck.go b/internal/scheduler/submitcheck.go
@@ -11,7 +11,6 @@ import (
 	"k8s.io/utils/clock"
 
 	"github.com/armadaproject/armada/internal/common/armadacontext"
-	"github.com/armadaproject/armada/internal/common/logging"
 	armadaslices "github.com/armadaproject/armada/internal/common/slices"
 	"github.com/armadaproject/armada/internal/scheduler/configuration"
 	"github.com/armadaproject/armada/internal/scheduler/database"
@@ -90,16 +89,16 @@ func (srv *SubmitChecker) Run(ctx *armadacontext.Context) error {
 func (srv *SubmitChecker) updateExecutors(ctx *armadacontext.Context) {
 	queues, err := srv.queueCache.GetAll(ctx)
 	if err != nil {
-		logging.
-			WithStacktrace(ctx, err).
+		ctx.Logger().
+			WithStacktrace(err).
 			Error("Error fetching queues")
 		return
 	}
 
 	executors, err := srv.executorRepository.GetExecutors(ctx)
 	if err != nil {
-		logging.
-			WithStacktrace(ctx, err).
+		ctx.Logger().
+			WithStacktrace(err).
 			Error("Error fetching executors")
 		return
 	}
@@ -140,8 +139,8 @@ func (srv *SubmitChecker) updateExecutors(ctx *armadacontext.Context) {
 					nodeDb: nodeDb,
 				}
 			} else {
-				logging.
-					WithStacktrace(ctx, err).
+				ctx.Logger().
+					WithStacktrace(err).
 					Warnf("Error constructing nodedb for executor: %s", ex.Id)
 			}
 		}

diff --git a/internal/scheduleringester/ingester.go b/internal/scheduleringester/ingester.go
@@ -15,7 +15,7 @@ import (
 	controlplaneevents_ingest_utils "github.com/armadaproject/armada/internal/common/ingest/controlplaneevents"
 	"github.com/armadaproject/armada/internal/common/ingest/jobsetevents"
 	"github.com/armadaproject/armada/internal/common/ingest/metrics"
-	log "github.com/armadaproject/armada/internal/common/loggin
+	log "github.com/armadaproject/armada/internal/common/logging"
 	"github.com/armadaproject/armada/internal/common/profiling"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 	"github.com/armadaproject/armada/pkg/controlplaneevents"