Merge branch 'main' into frrist/fix/datadir-path

cmd/cli/devstack/devstack.go

@@ -9,6 +9,7 @@ import (
 	"github.com/spf13/viper"
 	"k8s.io/kubectl/pkg/util/i18n"
 
+	"github.com/bacalhau-project/bacalhau/cmd/util/flags/cliflags"
 	"github.com/bacalhau-project/bacalhau/cmd/util/flags/configflags"
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 	"github.com/bacalhau-project/bacalhau/pkg/config_legacy"
@@ -82,7 +83,6 @@ func NewCmd() *cobra.Command {
 		RunE: func(cmd *cobra.Command, _ []string) error {
 			// ensure we either use a temp repo for the devstack, or the repo path provided
 			// by the specific devstack flag. Never use the default bacalhau repo.
-			v := viper.GetViper()
 			repoPath := ODs.ConfigurationRepo
 			if repoPath == "" {
 				// We need to clean up the repo when the node shuts down, but we can ONLY
@@ -94,8 +94,12 @@ func NewCmd() *cobra.Command {
 				// a different path to the one we've just created. Presumably a default.
 				defer os.RemoveAll(repoPath)
 			}
+
+			v := viper.GetViper()
+			configMap := v.GetStringMap(cliflags.RootCommandConfigValues)
+			configMap[types.DataDirKey] = repoPath
+			v.Set(cliflags.RootCommandConfigValues, configMap)
 			// override the repo path set in the root command with the derived path.
-			v.Set("repo", repoPath)
 			cfg, err := util.SetupConfig(cmd)
 			if err != nil {
 				return fmt.Errorf("setting up config: %w", err)

pkg/config/types/default_config.go

@@ -8,8 +8,6 @@ import (
 
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
-
-	"github.com/bacalhau-project/bacalhau/pkg/publisher/local"
 )
 
 const (
@@ -73,9 +71,6 @@ var Default = Bacalhau{
 					CPU:    "500m",
 					Memory: "1Gb",
 				},
-				Publisher: DefaultPublisherConfig{
-					Config: *local.NewSpecConfig(),
-				},
 			},
 		},
 		Ops: BatchJobDefaultsConfig{
@@ -85,9 +80,6 @@ var Default = Bacalhau{
 					CPU:    "500m",
 					Memory: "1Gb",
 				},
-				Publisher: DefaultPublisherConfig{
-					Config: *local.NewSpecConfig(),
-				},
 			},
 		},
 		Daemon: LongRunningJobDefaultsConfig{

pkg/jobstore/boltdb/store.go

@@ -662,6 +662,12 @@ func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string, query jobstore.J
 
 	bkt, err := NewBucketPath(BucketJobs, jobID, BucketJobHistory).Get(tx, false)
 	if err != nil {
+		// If the bucket doesn't exist, then we return an empty response to maintain compatibility
+		// with < v1.5.0 versions as the history bucket name was renamed in v1.5.0 without migration
+		// as migration is not worth the complexity
+		if errors.Is(err, bolt.ErrBucketNotFound) {
+			return &jobstore.JobHistoryQueryResponse{}, nil
+		}
 		return nil, NewBoltDbError(err)
 	}
 

pkg/node/config_defaults.go

@@ -1,16 +1,18 @@
 package node
 
 import (
+	"flag"
+	"os"
 	"path"
 	"runtime"
+	"strings"
 	"time"
 
 	"github.com/bacalhau-project/bacalhau/pkg/bidstrategy/semantic"
 	compute_system "github.com/bacalhau-project/bacalhau/pkg/compute/capacity/system"
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 	legacy_types "github.com/bacalhau-project/bacalhau/pkg/config_legacy/types"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
 )
 
 func NewDefaultComputeParam(storagePath string) ComputeConfigParams {
@@ -113,7 +115,10 @@ var TestRequesterConfig = RequesterConfigParams{
 }
 
 func getRequesterConfigParams() RequesterConfigParams {
-	if system.GetEnvironment() == system.EnvironmentTest {
+	if strings.Contains(os.Args[0], "/_test/") ||
+		strings.HasSuffix(os.Args[0], ".test") ||
+		flag.Lookup("test.v") != nil ||
+		flag.Lookup("test.run") != nil {
 		return TestRequesterConfig
 	}
 	return DefaultRequesterConfig

pkg/publicapi/endpoint/orchestrator/job.go

@@ -97,10 +97,10 @@ func (e *Endpoint) getJob(c echo.Context) error { //nolint: gocyclo
 				continue
 			}
 			jobHistoryQueryResponse, err := e.store.GetJobHistory(ctx, jobID, jobstore.JobHistoryQuery{})
-			history := jobHistoryQueryResponse.JobHistory
 			if err != nil {
 				return err
 			}
+			history := jobHistoryQueryResponse.JobHistory
 			response.History = &apimodels.ListJobHistoryResponse{
 				Items: make([]*models.JobHistory, len(history)),
 			}

pkg/repo/fs.go

@@ -106,6 +106,9 @@ func (fsr *FsRepo) Init() error {
 	if err := fsr.WriteVersion(Version4); err != nil {
 		return fmt.Errorf("failed to persist repo version: %w", err)
 	}
+	if err := fsr.WriteLegacyVersion(Version4); err != nil {
+		return fmt.Errorf("failed to persist legacy repo version: %w", err)
+	}
 
 	return nil
 }

pkg/telemetry/tracer.go

@@ -10,7 +10,6 @@ import (
 	oteltrace "go.opentelemetry.io/otel/trace"
 
 	_ "github.com/bacalhau-project/bacalhau/pkg/logger"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
 )
 
 // ----------------------------------------
@@ -33,23 +32,11 @@ func NewSpan(ctx context.Context, t oteltrace.Tracer, name string, opts ...otelt
 			))
 		}
 	}
-	opts = append(opts, oteltrace.WithAttributes(
-		attribute.String("environment", system.GetEnvironment().String()),
-	))
-
 	return t.Start(ctx, name, opts...)
 }
 
 func NewRootSpan(ctx context.Context, t oteltrace.Tracer, name string) (context.Context, oteltrace.Span) {
-	// Always include environment info in spans:
-	environment := system.GetEnvironment().String()
-	m0, _ := baggage.NewMember("environment", environment)
-	b, _ := baggage.New(m0)
-	ctx = baggage.ContextWithBaggage(ctx, b)
-
-	return t.Start(ctx, name, oteltrace.WithAttributes(
-		attribute.String("environment", environment),
-	))
+	return t.Start(ctx, name)
 }
 
 // Span creates and starts a new span, and a context containing it.
@@ -61,11 +48,6 @@ func NewRootSpan(ctx context.Context, t oteltrace.Tracer, name string) (context.
 // spanName: the name of the span, inside the service
 // opts: additional options to configure the span from trace.SpanStartOption
 func Span(ctx context.Context, spanName string, opts ...oteltrace.SpanStartOption) (context.Context, oteltrace.Span) {
-	// Always include environment info in spans:
-	opts = append(opts, oteltrace.WithAttributes(
-		attribute.String("environment", system.GetEnvironment().String()),
-	))
-
 	return GetTracer().Start(ctx, spanName, opts...)
 }
 

pkg/test/devstack/default_publisher_test.go

@@ -26,11 +26,6 @@ func TestDefaultPublisherSuite(t *testing.T) {
 }
 
 func (s *DefaultPublisherSuite) TestNoDefaultPublisher() {
-	s.Config.JobDefaults.Batch.Task.Publisher.Config.Type = ""
-	s.Config.JobDefaults.Batch.Task.Publisher.Config.Params = make(map[string]interface{})
-	stack := &scenario.StackConfig{}
-	stack.RequesterConfig.JobDefaults.Batch.Task.Publisher.Config.Type = ""
-	stack.RequesterConfig.JobDefaults.Batch.Task.Publisher.Config.Params = make(map[string]interface{})
 	testcase := scenario.Scenario{
 		Job: &models.Job{
 			Name:  s.T().Name(),
@@ -49,7 +44,6 @@ func (s *DefaultPublisherSuite) TestNoDefaultPublisher() {
 				},
 			},
 		},
-		Stack:          stack,
 		ResultsChecker: expectResultsNone,
 		JobCheckers: []scenario.StateChecks{
 			scenario.WaitForSuccessfulCompletion(),