events.go

package events

import (
	"context"
	stderrors "errors"
	"expvar"
	"io"
	"log"
	"time"

	"github.com/pkg/errors"

	"github.com/jeffrom/logd/config"
	"github.com/jeffrom/logd/internal"
	"github.com/jeffrom/logd/protocol"
	"github.com/jeffrom/logd/stats"
)

// this file contains the core logic of the program. Commands come from the
// various inputs. They are handled and a response is given. For example, a
// message is received, it is written to a backend, and a log id is returned to
// the caller.

var errInvalidFormat = stderrors.New("Invalid command format")

const defaultTopic = "default"

type flushState struct {
	conf    *config.Config
	batches int
	timer   *time.Timer
}

func newFlushState(conf *config.Config) *flushState {
	s := &flushState{
		conf: conf,
	}
	if conf.FlushInterval > 0 {
		s.timer = time.NewTimer(conf.FlushInterval)
	}
	return s
}

func (s *flushState) incr() {
	if s.conf.FlushBatches > 0 {
		s.batches++
	}
}

func (s *flushState) update() {
	if s.conf.FlushBatches > 0 && s.batches >= s.conf.FlushBatches {
		s.batches = 0
	}
}

func (s *flushState) shouldFlush() bool {
	if s.conf.FlushBatches > 0 {
		if s.batches >= s.conf.FlushBatches {
			return true
		}
	}
	if s.conf.FlushInterval > 0 {
		select {
		case <-s.timer.C:
			s.timer.Reset(s.conf.FlushInterval)
			return true
		default:
		}
	}
	return false
}

// eventQ synchronizes access to the log.
type eventQ struct {
	conf         *config.Config
	in           chan *protocol.Request
	stopC        chan error
	shutdownC    chan error
	topic        *topic
	partArgBuf   *partitionArgList
	batchScanner *protocol.BatchScanner
	Stats        *internal.Stats
	tmpBatch     *protocol.Batch
	flushState   *flushState
	confResp     *protocol.ConfigResponse
}

// newEventQ creates a new instance of an EventQ
func newEventQ(conf *config.Config) *eventQ {
	q := &eventQ{
		conf:         conf,
		Stats:        internal.NewStats(),
		in:           make(chan *protocol.Request, 1000),
		stopC:        make(chan error),
		shutdownC:    make(chan error, 1),
		partArgBuf:   newPartitionArgList(conf.MaxPartitions),
		batchScanner: protocol.NewBatchScanner(conf, nil),
		tmpBatch:     protocol.NewBatch(conf),
		flushState:   newFlushState(conf),
		confResp:     protocol.NewConfigResponse(conf),
	}

	return q
}

func (q *eventQ) setTopic(t *topic) {
	q.topic = t
}

// GoStart begins handling messages
func (q *eventQ) GoStart() error {
	go q.loop()
	return nil
}

// Start begins handling messages, blocking until the application is closed
func (q *eventQ) Start() error {
	if err := q.GoStart(); err != nil {
		return err
	}

	select {
	case err := <-q.shutdownC:
		if err != nil {
			return err
		}
	}
	return nil
}

func (q *eventQ) drainShutdownC() {
	for {
		select {
		case <-q.shutdownC:
		default:
			return
		}
	}
}

func (q *eventQ) loop() { // nolint: gocyclo
	q.drainShutdownC()
	defer func() {
		q.shutdownC <- nil
	}()

	for {
		internal.Debugf(q.conf, "waiting for event (%s)", q.topic)

		select {
		// new flow for handling requests passed in from servers
		case req := <-q.in:
			resp, err := q.handleRequest(req)

			if err != nil && err != protocol.ErrNotFound {
				log.Printf("error handling %s request: %+v", &req.Name, err)
			}
			req.Respond(resp)
		case <-q.stopC:
			return
		}
	}
}

// Stop halts the event queue
func (q *eventQ) Stop() error {
	var err error

	select {
	case q.stopC <- err:
	case <-time.After(500 * time.Millisecond):
		log.Printf("event queue failed to stop properly")
		return errors.New("shutdown timed out")
	}

	return nil
}

func (q *eventQ) handleBatch(req *protocol.Request) (*protocol.Response, error) {
	resp := req.Response
	q.tmpBatch.Reset()
	batch, err := q.tmpBatch.FromRequest(req)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	topic := q.topic
	if topic == nil {
		return errResponse(q.conf, req, resp, protocol.ErrNotFound)
	}

	// set next write partition if needed
	if topic.parts.shouldRotate(req.FullSize()) {
		if err := topic.idx.writeIndex(topic.parts.head.startOffset); err != nil {
			return errResponse(q.conf, req, resp, err)
		}

		nextStartOffset := topic.parts.nextOffset()
		if sperr := topic.logw.SetPartition(nextStartOffset); sperr != nil {
			return errResponse(q.conf, req, resp, sperr)
		}
	}
	// write the log
	_, err = topic.logw.Write(req.Bytes())
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	// maybe flush
	if ferr := q.doFlush(); ferr != nil {
		return errResponse(q.conf, req, resp, ferr)
	}

	// get response offset and update log state
	respOffset := topic.parts.nextOffset()
	_, aerr := topic.parts.addBatch(batch, req.FullSize())
	if aerr != nil {
		return errResponse(q.conf, req, resp, aerr)
	}

	// update query index
	if err := topic.Push(respOffset, topic.parts.head.startOffset, req.FullSize(), batch.Messages); err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	// respond
	cr := req.Response.ClientResponse
	cr.SetOffset(respOffset)
	cr.SetBatches(1)
	_, err = req.WriteResponse(resp, cr)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	return resp, nil
}

func (q *eventQ) doFlush() error {
	q.flushState.incr()
	if q.flushState.shouldFlush() {
		internal.Debugf(q.conf, "flushing topic %s", q.topic.name)
		if err := q.topic.logw.Flush(); err != nil {
			return err
		}
	}
	q.flushState.update()
	return nil
}

func (q *eventQ) handleRead(req *protocol.Request) (*protocol.Response, error) {
	resp := req.Response
	readreq, err := protocol.NewRead(q.conf).FromRequest(req)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	if readreq.Messages > q.conf.MaxMessages {
		return errResponse(q.conf, req, resp, protocol.ErrInvalid)
	}

	topic := q.topic
	if topic == nil {
		return errResponse(q.conf, req, resp, protocol.ErrNotFound)
	}

	args := partitionArgListPool.Get().(*partitionArgList).initialize(q.conf.MaxPartitions)
	defer partitionArgListPool.Put(args)
	partArgs, err := q.gatherReadArgs(topic, readreq.Offset, readreq.Messages, args)
	if err != nil {
		// fmt.Println("gatherReadArgs error:", err)

		// TODO test this. When the offset pointing to the very end of the file
		// is requested (which happens often when reading forever), we get
		// io.ErrUnexpectedEOF
		if err == io.ErrUnexpectedEOF {
			return errResponse(q.conf, req, resp, protocol.ErrNotFound)
		}
		return errResponse(q.conf, req, resp, protocol.ErrNotFound)
	}

	// respond OK
	cr := req.Response.ClientResponse
	cr.SetOffset(readreq.Offset)
	cr.SetBatches(partArgs.nbatches)
	// log.Panicf("%s %d", partArgs, partArgs.nbatches)
	_, err = req.WriteResponse(resp, cr)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	// respond with the batch(es)
	for i := 0; i < partArgs.nparts; i++ {
		args := partArgs.parts[i]
		p, gerr := topic.parts.logp.Get(args.offset, args.delta, args.limit)
		if gerr != nil {
			return errResponse(q.conf, req, resp, gerr)
		}

		if aerr := resp.AddReader(p); aerr != nil {
			return errResponse(q.conf, req, resp, aerr)
		}
	}

	return resp, nil
}

func (q *eventQ) handleTail(req *protocol.Request) (*protocol.Response, error) {
	resp := req.Response
	tailreq, err := protocol.NewTail(q.conf).FromRequest(req)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	if tailreq.Messages > q.conf.MaxMessages {
		return errResponse(q.conf, req, resp, protocol.ErrInvalid)
	}

	topic := q.topic
	if topic == nil {
		return errResponse(q.conf, req, resp, protocol.ErrNotFound)
	}

	firstPart := topic.parts.parts[0]
	if firstPart.size <= 0 {
		return errResponse(q.conf, req, resp, protocol.ErrNotFound)
	}
	off := firstPart.startOffset

	args := partitionArgListPool.Get().(*partitionArgList).initialize(q.conf.MaxPartitions)
	defer partitionArgListPool.Put(args)
	partArgs, err := q.gatherReadArgs(topic, off, tailreq.Messages, args)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	// respond OK
	cr := req.Response.ClientResponse
	cr.SetOffset(off)
	cr.SetBatches(partArgs.nbatches)
	_, err = req.WriteResponse(resp, cr)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}

	// respond with the batch(es)
	for i := 0; i < partArgs.nparts; i++ {
		args := partArgs.parts[i]
		p, gerr := topic.parts.logp.Get(args.offset, args.delta, args.limit)
		if gerr != nil {
			return errResponse(q.conf, req, resp, gerr)
		}

		if aerr := resp.AddReader(p); aerr != nil {
			return errResponse(q.conf, req, resp, aerr)
		}
	}
	return resp, nil
}

func (q *eventQ) handleStats(req *protocol.Request) (*protocol.Response, error) {
	resp := req.Response
	cr := req.Response.ClientResponse
	cr.SetMultiResp(stats.MultiOK())
	_, err := req.WriteResponse(resp, cr)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}
	return resp, nil
}

func (q *eventQ) handleClose(req *protocol.Request) (*protocol.Response, error) {
	resp := req.Response
	cr := req.Response.ClientResponse
	cr.SetOK()
	_, err := req.WriteResponse(resp, cr)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}
	return resp, nil
}

func (q *eventQ) handleConfig(req *protocol.Request) (*protocol.Response, error) {
	resp := req.Response
	cr := req.Response.ClientResponse
	cr.SetMultiResp(q.confResp.MultiResponse())
	_, err := req.WriteResponse(resp, cr)
	if err != nil {
		return errResponse(q.conf, req, resp, err)
	}
	return resp, nil
}

func (q *eventQ) gatherReadArgs(topic *topic, offset uint64, messages int, newArgs *partitionArgList) (*partitionArgList, error) {
	err := topic.idx.Query(offset, messages, newArgs)
	return newArgs, err
}

// handleShutdown handles a shutdown request
// TODO remove this
func (q *eventQ) handleShutdown() error {
	// check if shutdown command is allowed and wait to finish any outstanding
	// work here
	// TODO try all shutdowns or give up after the first error?
	return nil
}

// TODO maybe conns can just run this in their goroutine for nonblocking requests
func (q *eventQ) handleRequest(req *protocol.Request) (*protocol.Response, error) {
	var resp *protocol.Response
	var err error
	internal.Debugf(q.conf, "request: %s", &req.Name)

	switch req.Name {
	case protocol.CmdBatch:
		resp, err = q.handleBatch(req)
		instrumentRequest(stats.BatchRequests, stats.BatchErrors, err)
	case protocol.CmdRead:
		resp, err = q.handleRead(req)
		instrumentRequest(stats.ReadRequests, stats.ReadErrors, err)
	case protocol.CmdTail:
		resp, err = q.handleTail(req)
		instrumentRequest(stats.TailRequests, stats.TailErrors, err)
	case protocol.CmdStats:
		resp, err = q.handleStats(req)
		instrumentRequest(stats.StatsRequests, stats.StatsErrors, err)
	case protocol.CmdClose:
		resp, err = q.handleClose(req)
		instrumentRequest(stats.CloseRequests, stats.CloseErrors, err)
	case protocol.CmdConfig:
		resp, err = q.handleConfig(req)
		instrumentRequest(stats.ConfigRequests, stats.ConfigErrors, err)
	default:
		log.Printf("unhandled request type passed: %v", req.Name)
		resp = req.Response
		cr := req.Response.ClientResponse
		cr.SetError(protocol.ErrInvalid)
		err = protocol.ErrInvalid
		if _, werr := req.WriteResponse(resp, cr); werr != nil {
			err = werr
		}
	}

	return resp, err
}

// PushRequest adds a request event to the queue, and waits for a response.
// Called by server conn goroutines.
func (q *eventQ) PushRequest(ctx context.Context, req *protocol.Request) (*protocol.Response, error) {
	select {
	case q.in <- req:
	case <-ctx.Done():
		internal.Debugf(q.conf, "request %s cancelled", req)
		return nil, errors.New("request cancelled")
	}

	select {
	case resp := <-req.Responded():
		return resp, nil
	case <-ctx.Done():
		internal.Debugf(q.conf, "request %s cancelled while waiting for a response", req)
		return nil, errors.New("request cancelled")
	}
}

func errResponse(conf *config.Config, req *protocol.Request, resp *protocol.Response, err error) (*protocol.Response, error) {
	clientResp := req.Response.ClientResponse
	clientResp.SetError(err)
	if _, werr := req.WriteResponse(resp, clientResp); werr != nil {
		return resp, werr
	}
	return resp, err
}

func instrumentRequest(stat *expvar.Int, errStat *expvar.Int, err error) {
	stats.TotalRequests.Add(1)
	if err != nil {
		errStat.Add(1)
	} else {
		stat.Add(1)
	}
}