Reafactor schedules and engine

.travis.yml

@@ -1,7 +1,6 @@
 language: go
 go: 
-  - 1.7.3
-  - 1.8
+  - 1.8.3
   - tip
 
 go_import_path: github.com/yandex/pandora
@@ -13,6 +12,6 @@ before_install:
 
 install:
   - make tools
-  - go get -t `glide novendor` # Required to make vendoring in different PRs.
+  - go get -t `glide novendor`
 
 script: make travis

acceptance_tests/acceptance_suite_test.go

@@ -0,0 +1,159 @@
+package acceptance
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/ghodss/yaml"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/gbytes"
+	"github.com/onsi/gomega/gexec"
+	"go.uber.org/zap"
+
+	"github.com/onsi/gomega/format"
+	"github.com/yandex/pandora/lib/tag"
+	"github.com/yandex/pandora/lib/testutil"
+)
+
+var pandoraBin string
+
+func TestAcceptanceTests(t *testing.T) {
+	RegisterFailHandler(Fail)
+	format.UseStringerRepresentation = true
+
+	testutil.ReplaceGlobalLogger()
+	var args []string
+	if tag.Race {
+		zap.L().Debug("Building with race detector")
+		args = append(args, "-race")
+	}
+	if tag.Debug {
+		zap.L().Debug("Building with debug tag")
+		args = append(args, "-tags", "debug")
+	}
+	var err error
+	pandoraBin, err = gexec.Build("github.com/yandex/pandora", args...)
+	Expect(err).ToNot(HaveOccurred())
+	defer gexec.CleanupBuildArtifacts()
+	RunSpecs(t, "AcceptanceTests Suite")
+}
+
+type TestConfig struct {
+	// Default way to pass config to pandora.
+	PandoraConfig
+	// RawConfig overrides Pandora.
+	RawConfig  string
+	ConfigName string            // Without extension. "load" by default.
+	UseJSON    bool              // Using YAML by default.
+	CmdArgs    []string          // Nothing by default.
+	Files      map[string]string // Extra files to put in dir. Ammo, etc.
+}
+
+func NewTestConfig() *TestConfig {
+	return &TestConfig{
+		PandoraConfig: PandoraConfig{
+			Pool: []*InstancePoolConfig{NewInstansePoolConfig()},
+		},
+		Files: map[string]string{},
+	}
+}
+
+// PandoraConfig will be encoded to file via github.com/ghodss/yaml that supports json tags.
+// Or it can be encoded as JSON too, to test pandora JSON support.
+type PandoraConfig struct {
+	Pool []*InstancePoolConfig `json:"pools"`
+}
+
+func (pc *PandoraConfig) Append(ipc *InstancePoolConfig) {
+	pc.Pool = append(pc.Pool, ipc)
+}
+
+func NewInstansePoolConfig() *InstancePoolConfig {
+	return &InstancePoolConfig{
+		Provider:        map[string]interface{}{},
+		Aggregator:      map[string]interface{}{},
+		Gun:             map[string]interface{}{},
+		RPSSchedule:     map[string]interface{}{},
+		StartupSchedule: map[string]interface{}{},
+	}
+
+}
+
+type InstancePoolConfig struct {
+	Id              string
+	Provider        map[string]interface{} `json:"ammo"`
+	Aggregator      map[string]interface{} `json:"result"`
+	Gun             map[string]interface{} `json:"gun"`
+	RPSPerInstance  bool                   `json:"rps-per-instance"`
+	RPSSchedule     interface{}            `json:"rps"`
+	StartupSchedule interface{}            `json:"startup"`
+}
+
+type PandoraTester struct {
+	*gexec.Session
+	// TestDir is working dir of launched pandora.
+	// It contains config and ammo files, and will be removed after test execution.
+	// All files created during a test should created in this dir.
+	TestDir string
+	Config  *TestConfig
+}
+
+func NewTester(conf *TestConfig) *PandoraTester {
+	testDir, err := ioutil.TempDir("", "pandora_acceptance_")
+	Expect(err).ToNot(HaveOccurred())
+	if conf.ConfigName == "" {
+		conf.ConfigName = "load"
+	}
+	extension := "yaml"
+	if conf.UseJSON {
+		extension = "json"
+	}
+	var confData []byte
+
+	if conf.RawConfig != "" {
+		confData = []byte(conf.RawConfig)
+	} else {
+		if conf.UseJSON {
+			confData, err = json.Marshal(conf.PandoraConfig)
+		} else {
+			confData, err = yaml.Marshal(conf.PandoraConfig)
+		}
+		Expect(err).ToNot(HaveOccurred())
+	}
+	confAbsName := filepath.Join(testDir, conf.ConfigName+"."+extension)
+	err = ioutil.WriteFile(confAbsName, confData, 0644)
+	Expect(err).ToNot(HaveOccurred())
+
+	for file, data := range conf.Files {
+		fileAbsName := filepath.Join(testDir, file)
+		err = ioutil.WriteFile(fileAbsName, []byte(data), 0644)
+		Expect(err).ToNot(HaveOccurred())
+	}
+
+	command := exec.Command(pandoraBin, conf.CmdArgs...)
+	command.Dir = testDir
+	session, err := gexec.Start(command, GinkgoWriter, GinkgoWriter)
+	Expect(err).ToNot(HaveOccurred())
+
+	tt := &PandoraTester{
+		Session: session,
+		TestDir: testDir,
+		Config:  conf,
+	}
+	return tt
+}
+
+func (pt *PandoraTester) ShouldSay(pattern string) {
+	EventuallyWithOffset(1, pt.Out, 3*time.Second).Should(gbytes.Say(pattern))
+}
+
+func (pt *PandoraTester) Close() {
+	pt.Terminate()
+	os.RemoveAll(pt.TestDir)
+}

acceptance_tests/http_test.go

@@ -0,0 +1,82 @@
+package acceptance
+
+import (
+	"net/http"
+	"net/http/httptest"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"go.uber.org/atomic"
+)
+
+var _ = Describe("http", func() {
+	var (
+		server *httptest.Server
+		conf   *TestConfig
+		tester *PandoraTester
+	)
+	ServerEndpoint := func() string {
+		Expect(server).NotTo(BeNil())
+		return server.Listener.Addr().String()
+	}
+	BeforeEach(func() {
+		conf = NewTestConfig()
+		tester = nil
+	})
+	JustBeforeEach(func() {
+		tester = NewTester(conf)
+	})
+	AfterEach(func() {
+		tester.Close()
+		if server != nil {
+			server.Close()
+			server = nil
+		}
+	})
+
+	Context("jsonline", func() {
+		var (
+			requetsCount atomic.Int64
+		)
+		const (
+			Requests  = 4
+			Instances = 2
+			OutFile   = "out.log"
+		)
+		BeforeEach(func() {
+			server = httptest.NewServer(http.HandlerFunc(
+				func(rw http.ResponseWriter, req *http.Request) {
+					requetsCount.Inc()
+					rw.WriteHeader(http.StatusOK)
+				}))
+
+			conf.Pool[0].Gun = map[string]interface{}{
+				"type":   "http",
+				"target": ServerEndpoint(),
+			}
+			const ammoFile = "ammo.uri"
+			conf.Pool[0].Provider = map[string]interface{}{
+				"type": "uri",
+				"file": ammoFile,
+			}
+			conf.Files[ammoFile] = "/"
+			conf.Pool[0].Aggregator = map[string]interface{}{
+				"type":        "phout",
+				"destination": OutFile,
+			}
+			conf.Pool[0].RPSSchedule = []map[string]interface{}{
+				{"type": "once", "times": Requests / 2},
+				{"type": "const", "ops": Requests, "duration": "0.5s"},
+			}
+			conf.Pool[0].StartupSchedule = []map[string]interface{}{
+				{"type": "once", "times": Instances},
+			}
+		})
+		It("", func() {
+			exitCode := tester.Session.Wait(5).ExitCode()
+			Expect(exitCode).To(BeZero())
+			Expect(requetsCount.Load()).To(BeEquivalentTo(Requests))
+		})
+	})
+
+})