Merge pull request #30 from vmarkovtsev/master

Add hercules-combine

Makefile

@@ -2,7 +2,7 @@ ifneq (oneshell, $(findstring oneshell, $(.FEATURES)))
   $(error GNU make 3.82 or later is required)
 endif
 
-all: ${GOPATH}/bin/hercules ${GOPATH}/bin/hercules-generate-plugin
+all: ${GOPATH}/bin/hercules ${GOPATH}/bin/hercules-generate-plugin ${GOPATH}/bin/hercules-combine
 
 test: all
 	go test gopkg.in/src-d/hercules.v3
@@ -31,3 +31,6 @@ ${GOPATH}/bin/hercules: dependencies *.go cmd/hercules/*.go rbtree/*.go yaml/*.g
 
 ${GOPATH}/bin/hercules-generate-plugin: cmd/hercules-generate-plugin/*.go ${GOPATH}/bin/hercules
 	go get -ldflags "-X gopkg.in/src-d/hercules.v3.GIT_HASH=$$(git rev-parse HEAD)" gopkg.in/src-d/hercules.v3/cmd/hercules-generate-plugin
+
+${GOPATH}/bin/hercules-combine: cmd/hercules-combine/*.go ${GOPATH}/bin/hercules
+	go get -ldflags "-X gopkg.in/src-d/hercules.v3.GIT_HASH=$$(git rev-parse HEAD)" gopkg.in/src-d/hercules.v3/cmd/hercules-combine

cmd/hercules-combine/main.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"strings"
+
+	"github.com/gogo/protobuf/proto"
+	"gopkg.in/src-d/hercules.v3"
+	"gopkg.in/src-d/hercules.v3/pb"
+	"sort"
+)
+
+func main() {
+	files := os.Args[1:]
+	if len(files) == 0 {
+		fmt.Fprintln(os.Stderr, "Usage: hercules-combine file [file...]")
+		os.Exit(1)
+	}
+	if len(files) == 1 {
+		file, err := os.Open(files[0])
+		if err != nil {
+			panic(err)
+		}
+		defer file.Close()
+		io.Copy(os.Stdout, bufio.NewReader(file))
+		return
+	}
+	repos := []string{}
+	allErrors := map[string][]string{}
+	mergedResults := map[string]interface{}{}
+	mergedMetadata := &hercules.CommonAnalysisResult{}
+	for _, fileName := range files {
+		anotherResults, anotherMetadata, errs := loadMessage(fileName, &repos)
+		if anotherMetadata != nil {
+			mergeResults(mergedResults, mergedMetadata, anotherResults, anotherMetadata)
+		}
+		allErrors[fileName] = errs
+	}
+	printErrors(allErrors)
+	sort.Strings(repos)
+	if mergedMetadata == nil {
+		return
+	}
+	mergedMessage := pb.AnalysisResults{
+		Header:   &pb.Metadata{
+			Version:       2,
+			Hash:          hercules.GIT_HASH,
+			Repository:    strings.Join(repos, " & "),
+		},
+		Contents: map[string][]byte{},
+	}
+	mergedMetadata.FillMetadata(mergedMessage.Header)
+	for key, val := range mergedResults {
+		buffer := bytes.Buffer{}
+		hercules.Registry.Summon(key)[0].(hercules.LeafPipelineItem).Serialize(
+			val, true, &buffer)
+		mergedMessage.Contents[key] = buffer.Bytes()
+	}
+	serialized, err := proto.Marshal(&mergedMessage)
+	if err != nil {
+		panic(err)
+	}
+	os.Stdout.Write(serialized)
+}
+
+func loadMessage(fileName string, repos *[]string) (
+		map[string]interface{}, *hercules.CommonAnalysisResult, []string) {
+	errs := []string{}
+	buffer, err := ioutil.ReadFile(fileName)
+	if err != nil {
+		errs = append(errs, "Cannot read " + fileName + ": " + err.Error())
+		return nil, nil, errs
+	}
+	message := pb.AnalysisResults{}
+	err = proto.Unmarshal(buffer, &message)
+	if err != nil {
+		errs = append(errs, "Cannot parse " + fileName + ": " + err.Error())
+		return nil, nil, errs
+	}
+	*repos = append(*repos, message.Header.Repository)
+	results := map[string]interface{}{}
+	for key, val := range message.Contents {
+		summoned := hercules.Registry.Summon(key)
+		if len(summoned) == 0 {
+			errs = append(errs, fileName + ": item not found: " + key)
+			continue
+		}
+		mpi, ok := summoned[0].(hercules.MergeablePipelineItem)
+		if !ok {
+			errs = append(errs, fileName + ": " + key + ": MergeablePipelineItem is not implemented")
+			continue
+		}
+		msg, err := mpi.Deserialize(val)
+		if err != nil {
+			errs = append(errs, fileName + ": deserialization failed: " + key + ": " + err.Error())
+			continue
+		}
+		results[key] = msg
+	}
+	return results, hercules.MetadataToCommonAnalysisResult(message.Header), errs
+}
+
+func printErrors(allErrors map[string][]string) {
+	needToPrintErrors := false
+	for _, errs := range allErrors {
+		if len(errs) > 0 {
+			needToPrintErrors = true
+			break
+		}
+	}
+	if !needToPrintErrors {
+	 return
+	}
+	fmt.Fprintln(os.Stderr, "Errors:")
+	for key, errs := range allErrors {
+		if len(errs) > 0 {
+			fmt.Fprintln(os.Stderr, "  " + key)
+			for _, err := range errs {
+				fmt.Fprintln(os.Stderr, "    " + err)
+			}
+		}
+	}
+}
+
+func mergeResults(mergedResults map[string]interface{},
+		mergedCommons *hercules.CommonAnalysisResult,
+		anotherResults map[string]interface{},
+		anotherCommons *hercules.CommonAnalysisResult) {
+	for key, val := range anotherResults {
+		mergedResult, exists := mergedResults[key]
+		if !exists {
+			mergedResults[key] = val
+			continue
+		}
+		item := hercules.Registry.Summon(key)[0].(hercules.MergeablePipelineItem)
+		mergedResult, *mergedCommons = item.MergeResults(
+			mergedResult, val, mergedCommons, anotherCommons)
+		mergedResults[key] = mergedResult
+	}
+	if mergedCommons.CommitsNumber == 0 {
+		*mergedCommons = *anotherCommons
+	}
+}

cmd/hercules/main.go

@@ -257,7 +257,7 @@ func main() {
 func printResults(
 	uri string, deployed []hercules.LeafPipelineItem,
 	results map[hercules.LeafPipelineItem]interface{}) {
-	commonResult := results[nil].(hercules.CommonAnalysisResult)
+	commonResult := results[nil].(*hercules.CommonAnalysisResult)
 
 	fmt.Println("hercules:")
 	fmt.Println("  version: 3")
@@ -280,17 +280,13 @@ func printResults(
 func protobufResults(
 	uri string, deployed []hercules.LeafPipelineItem,
 	results map[hercules.LeafPipelineItem]interface{}) {
-	commonResult := results[nil].(hercules.CommonAnalysisResult)
 
 	header := pb.Metadata{
-		Version:       1,
-		Hash:          hercules.GIT_HASH,
-		Repository:    uri,
-		BeginUnixTime: commonResult.BeginTime,
-		EndUnixTime:   commonResult.EndTime,
-		Commits:       int32(commonResult.CommitsNumber),
-		RunTime:       commonResult.RunTime.Nanoseconds() / 1e6,
+		Version:    2,
+		Hash:       hercules.GIT_HASH,
+		Repository: uri,
 	}
+	results[nil].(*hercules.CommonAnalysisResult).FillMetadata(&header)
 
 	message := pb.AnalysisResults{
 		Header:   &header,

pipeline.go

@@ -19,6 +19,7 @@ import (
 	"gopkg.in/src-d/go-git.v4/plumbing"
 	"gopkg.in/src-d/go-git.v4/plumbing/object"
 	"gopkg.in/src-d/hercules.v3/toposort"
+	"gopkg.in/src-d/hercules.v3/pb"
 )
 
 type ConfigurationOptionType int
@@ -99,8 +100,8 @@ type MergeablePipelineItem interface {
 	LeafPipelineItem
 	// Deserialize loads the result from Protocol Buffers blob.
 	Deserialize(pbmessage []byte) (interface{}, error)
-	// MergeResults joins two results together.
-	MergeResults(r1, r2 interface{}, c1, c2 CommonAnalysisResult) interface{}
+	// MergeResults joins two results together. Common-s are specified as the global state.
+	MergeResults(r1, r2 interface{}, c1, c2 *CommonAnalysisResult) (interface{}, CommonAnalysisResult)
 }
 
 // CommonAnalysisResult holds the information which is always extracted at Pipeline.Run().
@@ -115,6 +116,22 @@ type CommonAnalysisResult struct {
 	RunTime time.Duration
 }
 
+func (car *CommonAnalysisResult) FillMetadata(meta *pb.Metadata) {
+	meta.BeginUnixTime = car.BeginTime
+	meta.EndUnixTime = car.EndTime
+	meta.Commits = int32(car.CommitsNumber)
+	meta.RunTime = car.RunTime.Nanoseconds() / 1e6
+}
+
+func MetadataToCommonAnalysisResult(meta *pb.Metadata) *CommonAnalysisResult {
+  return &CommonAnalysisResult{
+	  BeginTime:     meta.BeginUnixTime,
+	  EndTime:       meta.EndUnixTime,
+	  CommitsNumber: int(meta.Commits),
+	  RunTime:       time.Duration(meta.RunTime * 1e6),
+  }
+}
+
 // PipelineItemRegistry contains all the known PipelineItem-s.
 type PipelineItemRegistry struct {
 	provided   map[string][]reflect.Type
@@ -582,7 +599,7 @@ func (pipeline *Pipeline) Run(commits []*object.Commit) (map[LeafPipelineItem]in
 			result[casted] = casted.Finalize()
 		}
 	}
-	result[nil] = CommonAnalysisResult{
+	result[nil] = &CommonAnalysisResult{
 		BeginTime:     commits[0].Author.When.Unix(),
 		EndTime:       commits[len(commits)-1].Author.When.Unix(),
 		CommitsNumber: len(commits),

pipeline_test.go

@@ -217,7 +217,7 @@ func TestPipelineRun(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 2, len(result))
 	assert.Equal(t, item, result[item].(*testPipelineItem))
-	common := result[nil].(CommonAnalysisResult)
+	common := result[nil].(*CommonAnalysisResult)
 	assert.Equal(t, common.BeginTime, int64(1481719092))
 	assert.Equal(t, common.EndTime, int64(1481719092))
 	assert.Equal(t, common.CommitsNumber, 1)