<commit_before>package migration

import (
	"context"
	"fmt"
	"path"

	"migration/onefoureight/db/pfs"
	"migration/onefoureight/db/pps"

	"github.com/gogo/protobuf/proto"
	"github.com/pachyderm/pachyderm/src/client"

	etcd "github.com/coreos/etcd/clientv3"
	"go.pedge.io/lion/proto"
)

const (
	reposPrefix     = "/repos"
	commitsPrefix   = "/commits"
	branchesPrefix  = "/branches"
	pipelinesPrefix = "/pipelines"
	jobsPrefix      = "/jobs"
)

func oneFourToOneFive(etcdAddress, pfsPrefix, ppsPrefix string) error {
	etcdClient, err := etcd.New(etcd.Config{
		Endpoints:   []string{fmt.Sprintf("%s:2379", etcdAddress)},
		DialOptions: client.EtcdDialOptions(),
	})
	if err != nil {
		return fmt.Errorf("error constructing etcdClient: %v", err)
	}

	// This function migrates objects under a specific prefix
	migrate := func(prefix string, template proto.Message) {
		resp, err := etcdClient.Get(context.Background(), prefix, etcd.WithPrefix())
		if err != nil {
			protolion.Errorf("error getting %v: %v", prefix, err)
			return
		}
		for _, kv := range resp.Kvs {
			key := string(kv.Key)
			if err := proto.UnmarshalText(string(kv.Value), template); err != nil {
				protolion.Errorf("error unmarshalling object %v: %v", key, err)
				continue
			}
			bytes, err := proto.Marshal(template)
			if err != nil {
				protolion.Errorf("error marshalling object %v: %v", key, err)
				continue
			}
			if _, err := etcdClient.Put(context.Background(), key, string(bytes)); err != nil {
				protolion.Errorf("error putting object %v: %v", key, err)
				continue
			}
		}
	}

	var repoInfo pfs.RepoInfo
	migrate(path.Join(pfsPrefix, reposPrefix), &repoInfo)
	protolion.Infof("finished migrating repos")

	var commitInfo pfs.CommitInfo
	migrate(path.Join(pfsPrefix, commitsPrefix), &commitInfo)
	protolion.Infof("finished migrating commits")

	var head pfs.Commit
	migrate(path.Join(pfsPrefix, branchesPrefix), &head)
	protolion.Infof("finished migrating branches")

	var pipelineInfo pps.PipelineInfo
	migrate(path.Join(ppsPrefix, pipelinesPrefix), &pipelineInfo)
	protolion.Infof("finished migrating pipelines")

	var jobInfo pps.JobInfo
	migrate(path.Join(ppsPrefix, jobsPrefix), &jobInfo)
	protolion.Infof("finished migrating jobs")

	return nil
}
<commit_msg>Sort objects to migrate by timestamp<commit_after>package migration

import (
	"context"
	"fmt"
	"path"

	"migration/onefoureight/db/pfs"
	"migration/onefoureight/db/pps"

	"github.com/gogo/protobuf/proto"
	"github.com/pachyderm/pachyderm/src/client"

	etcd "github.com/coreos/etcd/clientv3"
	"go.pedge.io/lion/proto"
)

const (
	reposPrefix     = "/repos"
	commitsPrefix   = "/commits"
	branchesPrefix  = "/branches"
	pipelinesPrefix = "/pipelines"
	jobsPrefix      = "/jobs"
)

func oneFourToOneFive(etcdAddress, pfsPrefix, ppsPrefix string) error {
	etcdClient, err := etcd.New(etcd.Config{
		Endpoints:   []string{fmt.Sprintf("%s:2379", etcdAddress)},
		DialOptions: client.EtcdDialOptions(),
	})
	if err != nil {
		return fmt.Errorf("error constructing etcdClient: %v", err)
	}

	// This function migrates objects under a specific prefix
	migrate := func(prefix string, template proto.Message) {
		// We want to sort the objects by oldest-to-newest order,
		// so we preserve their timestamp ordering as we update them.
		resp, err := etcdClient.Get(context.Background(), prefix, etcd.WithPrefix(), etcd.WithSort(etcd.SortByModRevision, etcd.SortAscend))
		if err != nil {
			protolion.Errorf("error getting %v: %v", prefix, err)
			return
		}
		for _, kv := range resp.Kvs {
			key := string(kv.Key)
			if err := proto.UnmarshalText(string(kv.Value), template); err != nil {
				protolion.Errorf("error unmarshalling object %v: %v", key, err)
				continue
			}
			bytes, err := proto.Marshal(template)
			if err != nil {
				protolion.Errorf("error marshalling object %v: %v", key, err)
				continue
			}
			if _, err := etcdClient.Put(context.Background(), key, string(bytes)); err != nil {
				protolion.Errorf("error putting object %v: %v", key, err)
				continue
			}
		}
	}

	var repoInfo pfs.RepoInfo
	migrate(path.Join(pfsPrefix, reposPrefix), &repoInfo)
	protolion.Infof("finished migrating repos")

	var commitInfo pfs.CommitInfo
	migrate(path.Join(pfsPrefix, commitsPrefix), &commitInfo)
	protolion.Infof("finished migrating commits")

	var head pfs.Commit
	migrate(path.Join(pfsPrefix, branchesPrefix), &head)
	protolion.Infof("finished migrating branches")

	var pipelineInfo pps.PipelineInfo
	migrate(path.Join(ppsPrefix, pipelinesPrefix), &pipelineInfo)
	protolion.Infof("finished migrating pipelines")

	var jobInfo pps.JobInfo
	migrate(path.Join(ppsPrefix, jobsPrefix), &jobInfo)
	protolion.Infof("finished migrating jobs")

	return nil
}

{'fraction_non_alphanumeric': 0.13244331426149608, 'fraction_numerical': 0.0021190930281839372, 'mean_word_length': 4.444059976931949, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3863915', 'n_tokens_mistral': 1796, 'n_tokens_neox': 1634, 'n_words': 368}