<commit_before>//go:generate bitfanDoc
// Listen and read a http request to build events with it.
//
// Processor respond with a HTTP code as :
//
// * `202` when request has been accepted, in body : the total number of event created
// * `500` when an error occurs, in body : an error description
//
// Use codecs to process body content as json / csv / lines / json lines / ....
//
// URL is available as http://webhookhost/pluginLabel/URI
//
// * webhookhost is defined by bitfan at startup
// * pluginLabel is defined in pipeline configuration, it's the named processor if you put one, or `input_httpserver` by default
// * URI is defined in plugin configuration (see below)
package httpserverprocessor

import (
	"fmt"
	"io"
	"net/http"
	"net/http/httputil"
	"os"

	"github.com/vjeantet/bitfan/codecs"
	"github.com/vjeantet/bitfan/processors"
)

func New() processors.Processor {
	return &processor{opt: &options{}}
}

type options struct {
	processors.CommonOptions `mapstructure:",squash"`

	// The codec used for input data. Input codecs are a convenient method for decoding
	// your data before it enters the input, without needing a separate filter in your bitfan pipeline
	// @Default "plain"
	// @Type codec
	Codec codecs.Codec

	// URI path
	// @Default "events"
	Uri string
}

// Reads events from standard input
type processor struct {
	processors.Base

	opt  *options
	q    chan bool
	host string
}

func (p *processor) Configure(ctx processors.ProcessorContext, conf map[string]interface{}) error {
	defaults := options{
		Codec: codecs.New("plain", nil, ctx.Log(), ctx.ConfigWorkingLocation()),
		Uri:   "events",
	}
	p.opt = &defaults
	err := p.ConfigureAndValidate(ctx, conf, p.opt)
	if err != nil {
		return err
	}

	if p.host, err = os.Hostname(); err != nil {
		p.Logger.Warnf("can not get hostname : %s", err.Error())
	}

	return err
}
func (p *processor) Start(e processors.IPacket) error {
	p.q = make(chan bool)
	p.WebHook.Add(p.opt.Uri, p.HttpHandler)
	return nil
}

// Handle Request received by bitfan for this agent (url hook should be registered during p.Start)
func (p *processor) HttpHandler(w http.ResponseWriter, r *http.Request) {
	p.Logger.Debug("reading request")

	// Create a reader
	var dec codecs.Decoder
	var err error

	if dec, err = p.opt.Codec.NewDecoder(r.Body); err != nil {
		p.Logger.Errorln("decoder error : ", err.Error())
		return
	}
	headersBytes, _ := httputil.DumpRequest(r, false)
	headers := string(headersBytes)

	req := map[string]interface{}{
		"remoteAddr":  r.RemoteAddr,
		"rawHeaders":  headers,
		"method":      r.Method,
		"requestURI":  r.RequestURI,
		"proto":       r.Proto,
		"host":        r.Host,
		"requestPath": r.URL.Path,
	}

	req["querystring"] = map[string]interface{}{}
	for i, v := range r.URL.Query() {
		if len(v) == 1 {
			req["querystring"].(map[string]interface{})[i] = v[0]
		} else {
			req["querystring"].(map[string]interface{})[i] = v
		}
	}

	req["headers"] = map[string]interface{}{}
	for i, v := range r.Header {
		if len(v) == 1 {
			req["headers"].(map[string]interface{})[i] = v[0]
		} else {
			req["headers"].(map[string]interface{})[i] = v
		}
	}

	if r.Method == "POST" {
		r.ParseForm()
		req["formvalues"] = map[string]interface{}{}
		for i, v := range r.PostForm {
			if len(v) == 1 {
				req["formvalues"].(map[string]interface{})[i] = v[0]
			} else {
				req["formvalues"].(map[string]interface{})[i] = v
			}
		}
	}

	var nbEvents int
	p.Logger.Debug("request = ", req)
	p.Logger.Debug("start reading body content")

	for dec.More() {
		var record interface{}
		if err = dec.Decode(&record); err != nil {
			if err == io.EOF {
				p.Logger.Warnln("error while http read docoding : ", err)
			} else {
				p.Logger.Errorln("error while http read docoding : ", err)
				break
			}
		}

		var e processors.IPacket
		switch v := record.(type) {
		case nil:
			e = p.NewPacket("", map[string]interface{}{
				"request": req,
			})
		case string:
			e = p.NewPacket(v, map[string]interface{}{
				"request": req,
			})
		case map[string]interface{}:
			e = p.NewPacket("", v)
			e.Fields().SetValueForPath(req, "request")
		case []interface{}:
			e = p.NewPacket("", map[string]interface{}{
				"request": req,
				"data":    v,
			})
		default:
			p.Logger.Errorf("Unknow structure %#v", v)
		}

		p.opt.ProcessCommonOptions(e.Fields())
		p.Send(e)
		nbEvents++
		select {
		case <-p.q:
			return
		default:
		}
	}

	if err != nil && err != io.EOF {
		w.WriteHeader(http.StatusInternalServerError)
		w.Header().Set("Content-Type", "text/plain")
		w.Write([]byte("500 - " + err.Error()))
	} else {
		w.WriteHeader(http.StatusAccepted)
		w.Header().Set("Content-Type", "text/plain")
		w.Write([]byte(fmt.Sprintf("%d", nbEvents)))
	}

}

func (p *processor) Stop(e processors.IPacket) error {
	close(p.q)
	return nil
}<commit_msg>processor : httpserver - new param "headers" and ouput is now JSON with UIID of each generated events #30<commit_after>//go:generate bitfanDoc
// Listen and read a http request to build events with it.
//
// Processor respond with a HTTP code as :
//
// * `202` when request has been accepted, in body : the total number of event created
// * `500` when an error occurs, in body : an error description
//
// Use codecs to process body content as json / csv / lines / json lines / ....
//
// URL is available as http://webhookhost/pluginLabel/URI
//
// * webhookhost is defined by bitfan at startup
// * pluginLabel is defined in pipeline configuration, it's the named processor if you put one, or `input_httpserver` by default
// * URI is defined in plugin configuration (see below)
package httpserverprocessor

import (
	"io"
	"net/http"
	"net/http/httputil"
	"os"

	uuid "github.com/nu7hatch/gouuid"
	"github.com/vjeantet/bitfan/codecs"
	"github.com/vjeantet/bitfan/processors"
)

func New() processors.Processor {
	return &processor{opt: &options{}}
}

type options struct {
	processors.CommonOptions `mapstructure:",squash"`

	// The codec used for input data. Input codecs are a convenient method for decoding
	// your data before it enters the input, without needing a separate filter in your bitfan pipeline
	// @Default "plain"
	// @Type codec
	Codec codecs.Codec

	// URI path
	// @Default "events"
	Uri string

	// Headers to send back into each outgoing response
	// @LSExample {"X-Processor" => "bitfan"}
	Headers map[string]string `mapstructure:"headers"`

	// What to send back to client ?
	// @Default ["uuid"]
	Body []string `mapstructure:"body"`
}

// Reads events from standard input
type processor struct {
	processors.Base

	opt     *options
	q       chan bool
	host    string
	encoder codecs.Codec
}

func (p *processor) Configure(ctx processors.ProcessorContext, conf map[string]interface{}) error {
	defaults := options{
		Codec: codecs.New("plain", nil, ctx.Log(), ctx.ConfigWorkingLocation()),
		Uri:   "events",
		Body:  []string{"uuid"},
	}
	p.opt = &defaults
	err := p.ConfigureAndValidate(ctx, conf, p.opt)
	if err != nil {
		return err
	}

	if p.host, err = os.Hostname(); err != nil {
		p.Logger.Warnf("can not get hostname : %s", err.Error())
	}

	p.encoder = codecs.New("json", nil, ctx.Log(), ctx.ConfigWorkingLocation())
	return err
}
func (p *processor) Start(e processors.IPacket) error {
	p.q = make(chan bool)
	p.WebHook.Add(p.opt.Uri, p.HttpHandler)
	return nil
}

// Handle Request received by bitfan for this agent (url hook should be registered during p.Start)
func (p *processor) HttpHandler(w http.ResponseWriter, r *http.Request) {
	p.Logger.Debug("reading request")

	// Create a reader
	var dec codecs.Decoder
	var err error

	if dec, err = p.opt.Codec.NewDecoder(r.Body); err != nil {
		p.Logger.Errorln("decoder error : ", err.Error())
		return
	}
	headersBytes, _ := httputil.DumpRequest(r, false)
	headers := string(headersBytes)

	req := map[string]interface{}{
		"remoteAddr":  r.RemoteAddr,
		"rawHeaders":  headers,
		"method":      r.Method,
		"requestURI":  r.RequestURI,
		"proto":       r.Proto,
		"host":        r.Host,
		"requestPath": r.URL.Path,
	}

	req["querystring"] = map[string]interface{}{}
	for i, v := range r.URL.Query() {
		if len(v) == 1 {
			req["querystring"].(map[string]interface{})[i] = v[0]
		} else {
			req["querystring"].(map[string]interface{})[i] = v
		}
	}

	req["headers"] = map[string]interface{}{}
	for i, v := range r.Header {
		if len(v) == 1 {
			req["headers"].(map[string]interface{})[i] = v[0]
		} else {
			req["headers"].(map[string]interface{})[i] = v
		}
	}

	if r.Method == "POST" {
		r.ParseForm()
		req["formvalues"] = map[string]interface{}{}
		for i, v := range r.PostForm {
			if len(v) == 1 {
				req["formvalues"].(map[string]interface{})[i] = v[0]
			} else {
				req["formvalues"].(map[string]interface{})[i] = v
			}
		}
	}

	var responseData map[string]interface{}
	responseData = map[string]interface{}{}

	p.Logger.Debug("request = ", req)
	p.Logger.Debug("start reading body content")

	for dec.More() {
		var record interface{}
		var body map[string]interface{}
		body = map[string]interface{}{}
		if err = dec.Decode(&record); err != nil {
			if err == io.EOF {
				p.Logger.Warnln("error while http read docoding : ", err)
			} else {
				p.Logger.Errorln("error while http read docoding : ", err)
				break
			}
		}

		var e processors.IPacket
		switch v := record.(type) {
		case nil:
			e = p.NewPacket("", map[string]interface{}{
				"request": req,
			})
		case string:
			e = p.NewPacket(v, map[string]interface{}{
				"request": req,
			})
		case map[string]interface{}:
			e = p.NewPacket("", v)
			e.Fields().SetValueForPath(req, "request")
		case []interface{}:
			e = p.NewPacket("", map[string]interface{}{
				"request": req,
				"data":    v,
			})
		default:
			p.Logger.Errorf("Unknow structure %#v", v)
		}

		id, _ := uuid.NewV4()
		e.Fields().SetValueForPath(id.String(), "uuid")
		p.opt.ProcessCommonOptions(e.Fields())

		for _, path := range p.opt.Body {
			value, err := e.Fields().ValueForPath(path)
			if err != nil {
				p.Logger.Errorf("ValueForPath %s - %s", path, err)
				continue
			}
			body[path] = value
		}
		responseData[id.String()] = body

		p.Send(e)

		select {
		case <-p.q:
			return
		default:
		}
	}

	if err != nil && err != io.EOF {
		p.Logger.Errorln("error : ", err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		w.Write([]byte(err.Error()))
		return
	}

	// Encode content
	var enc codecs.Encoder
	enc, err = p.encoder.NewEncoder(w)
	if err != nil {
		p.Logger.Errorln("codec error : ", err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		w.Write([]byte(err.Error()))
		return
	}

	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	for hn, hv := range p.opt.Headers {
		w.Header().Set(hn, hv)
		p.Logger.Debugf("added header : %s -> %s", hn, hv)
	}
	w.WriteHeader(http.StatusAccepted)
	enc.Encode(responseData)
}

func (p *processor) Stop(e processors.IPacket) error {
	close(p.q)
	return nil
}
{'fraction_non_alphanumeric': 0.16144796380090498, 'fraction_numerical': 0.003076923076923077, 'mean_word_length': 3.5032599837000813, 'pattern_counts': {'":': 22, '<': 5, '<?xml version=': 0, '>': 5, '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': '4999712', 'n_tokens_mistral': 4171, 'n_tokens_neox': 3825, 'n_words': 1018}