hadock

README

the high rate data brocker AKA hadock has been developped to support high rate data activities in the frame of FSL operations at B.USOC.

main features of hadock are:

• processing of HRDL packets
  • storage in local archive
  • detection of invalid/corrupted packets
  • creation of processed parameters
  • cascading
• HTTP interface for clients to get raw or processed data archived by hadock
• monitoring of HRDL stream and hadock activities
• manual replay of RT files from HRDP archive.

Each of these features have a dedicated command built-in into hadock each summarized below .

the listen command implements the core of hadock: processing and storing incoming HRDL packets. this command expect that each HRDL packets are embedded in the hadock own protocol.

For each packets received, hadock will:

• identifiy the instance, mode, type of a packet,
• store the raw data in a dedicated directory with a configurable directory tree structure according to the values found in the headers of the HRDL packet,
• store metadata (XML format) next to the raw data when a packet is an image packet,
• summarize the status of the HRDL stream and generate processed parameters at regular interval for monitoring tools.

hadock can support multiple type of storage:

• file: data from HRDL packets are stored in individual file
• tar: data from HRDL packets are stored in tar archive that will be automatically rotated under certain conditions
• hrdp: each HRDL packets are stored in files similar to the RT files found in the HRDP archive.

the replay command has been initially written to develop and test the protocol used by hadock listen in order to process incoming HRDL packets.

Today, its usage has been extended to:

• validation of new feature of hadock,
• consolidation of hadock archives.

the distrib command runs a embedded HTTP server (that should be run behind a nginx or apache httpd server) to deliver data to clients (external tools or users).

This command used the raw archive of hadock as its source to deliver the data. For every request received, it will read the raw data and try to convert it into a readable format (eg: csv for sciences data, png|jpg for image data).

additional tools have been developped in the meantime and are available in their own dedicated repositories. These tools can be used for different purposes such as:

• scanning the hadock archive (upifinder)
• scanning the HRDP archive for VMU packets (meex)
• feeds hadock with HRDL packets from a stream of VCDU (erdle)
• process MVIS data (mvis2list)

Documentation

Index

• Constants
• Variables
• func DecodeBinaryPackets(r io.Reader, is []uint8) <-chan *Packet
• func DecodeCompressedPackets(r io.Reader, is []uint8) <-chan *Packet
• func DialProxy(a, e string) (io.WriteCloser, error)
• func EncodePacket(p *Packet) ([]byte, error)
• type Item
• type Message
  • func DecodeMessage(r io.Reader) (Message, error)
• type Module
  • func Process(ms []Module) Module
• type Notifier
  • func NewDebuggerNotifier(w io.Writer, o *Options) (Notifier, error)
  • func NewExternalNotifier(p, a string, o *Options) (Notifier, error)
• type Options
  • func (o *Options) Accept(msg Message) error
• type Packet
  • func DecodePacket(r io.Reader) (*Packet, error)
• type Pool
  • func NewPool(ns []Notifier, a, e time.Duration) *Pool
  • func (p *Pool) Notify(i *Item)

Constants

const (
	OPS  = 255
	TEST = 0
	SIM1 = 1
	SIM2 = 2
)

const (
	HadockVersion1 = iota
	HadockVersion2
)

const PacketLength = 10

const Preamble uint32 = 0xf82e3553

Variables

var (
	ErrUnsupportedProtocol   = errors.New("unsupported protocol")
	ErrUnsupportedVMUVersion = errors.New("unsupported vmu version")
	ErrSkip                  = errors.New("skip")
)

Functions

func DecodeBinaryPackets

func DecodeBinaryPackets(r io.Reader, is []uint8) <-chan *Packet

func DecodeCompressedPackets

func DecodeCompressedPackets(r io.Reader, is []uint8) <-chan *Packet

func DialProxy

func DialProxy(a, e string) (io.WriteCloser, error)

func EncodePacket

func EncodePacket(p *Packet) ([]byte, error)

Types

type Item

type Item struct {
	Instance int32
	panda.HRPacket
}

type Message

type Message struct {
	Origin    string        `json:"origin"`
	Sequence  uint32        `json:"sequence"`
	Instance  int32         `json:"instance"`
	Channel   panda.Channel `json:"channel"`
	Realtime  bool          `json:"realtime"`
	Count     uint32        `json:"count"`
	Elapsed   time.Duration `json:"elapsed"`
	Generated int64         `json:"generated"`
	Acquired  int64         `json:"acquired"`
	Reference string        `json:"reference"`
	UPI       string        `json:"upi"`
}

func DecodeMessage

func DecodeMessage(r io.Reader) (Message, error)

type Module

type Module interface {
	Process(uint8, panda.HRPacket) error
}

func Process

func Process(ms []Module) Module

type Notifier

type Notifier interface {
	Accept(Message) error
	Notify(Message) error
}

func NewDebuggerNotifier

func NewDebuggerNotifier(w io.Writer, o *Options) (Notifier, error)

func NewExternalNotifier

func NewExternalNotifier(p, a string, o *Options) (Notifier, error)

type Options

type Options struct {
	Source   string
	Instance int32
	Channels []panda.Channel
}

func (*Options) Accept

func (o *Options) Accept(msg Message) error

type Packet

type Packet struct {
	Protocol uint8
	Version  uint8
	Instance uint8
	Sequence uint16
	Length   uint32
	Payload  []byte
	Sum      uint16

	Curr uint16
	Last uint16
}

func DecodePacket

func DecodePacket(r io.Reader) (*Packet, error)

type Pool

type Pool struct {
	// contains filtered or unexported fields
}

func NewPool

func NewPool(ns []Notifier, a, e time.Duration) *Pool

func (*Pool) Notify

func (p *Pool) Notify(i *Item)