vendor/go.opencensus.io/stats/view/worker_commands.go

   1 // Copyright 2017, OpenCensus Authors
   2 //
   3 // Licensed under the Apache License, Version 2.0 (the "License");
   4 // you may not use this file except in compliance with the License.
   5 // You may obtain a copy of the License at
   6 //
   7 //     http://www.apache.org/licenses/LICENSE-2.0
   8 //
   9 // Unless required by applicable law or agreed to in writing, software
  10 // distributed under the License is distributed on an "AS IS" BASIS,
  11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 // See the License for the specific language governing permissions and
  13 // limitations under the License.
  14 //
  15
  16 package view
  17
  18 import (
  19         "errors"
  20         "fmt"
  21         "strings"
  22         "time"
  23
  24         "go.opencensus.io/exemplar"
  25
  26         "go.opencensus.io/stats"
  27         "go.opencensus.io/stats/internal"
  28         "go.opencensus.io/tag"
  29 )
  30
  31 type command interface {
  32         handleCommand(w *worker)
  33 }
  34
  35 // getViewByNameReq is the command to get a view given its name.
  36 type getViewByNameReq struct {
  37         name string
  38         c    chan *getViewByNameResp
  39 }
  40
  41 type getViewByNameResp struct {
  42         v *View
  43 }
  44
  45 func (cmd *getViewByNameReq) handleCommand(w *worker) {
  46         v := w.views[cmd.name]
  47         if v == nil {
  48                 cmd.c <- &getViewByNameResp{nil}
  49                 return
  50         }
  51         cmd.c <- &getViewByNameResp{v.view}
  52 }
  53
  54 // registerViewReq is the command to register a view.
  55 type registerViewReq struct {
  56         views []*View
  57         err   chan error
  58 }
  59
  60 func (cmd *registerViewReq) handleCommand(w *worker) {
  61         var errstr []string
  62         for _, view := range cmd.views {
  63                 vi, err := w.tryRegisterView(view)
  64                 if err != nil {
  65                         errstr = append(errstr, fmt.Sprintf("%s: %v", view.Name, err))
  66                         continue
  67                 }
  68                 internal.SubscriptionReporter(view.Measure.Name())
  69                 vi.subscribe()
  70         }
  71         if len(errstr) > 0 {
  72                 cmd.err <- errors.New(strings.Join(errstr, "\n"))
  73         } else {
  74                 cmd.err <- nil
  75         }
  76 }
  77
  78 // unregisterFromViewReq is the command to unregister to a view. Has no
  79 // impact on the data collection for client that are pulling data from the
  80 // library.
  81 type unregisterFromViewReq struct {
  82         views []string
  83         done  chan struct{}
  84 }
  85
  86 func (cmd *unregisterFromViewReq) handleCommand(w *worker) {
  87         for _, name := range cmd.views {
  88                 vi, ok := w.views[name]
  89                 if !ok {
  90                         continue
  91                 }
  92
  93                 // Report pending data for this view before removing it.
  94                 w.reportView(vi, time.Now())
  95
  96                 vi.unsubscribe()
  97                 if !vi.isSubscribed() {
  98                         // this was the last subscription and view is not collecting anymore.
  99                         // The collected data can be cleared.
 100                         vi.clearRows()
 101                 }
 102                 delete(w.views, name)
 103         }
 104         cmd.done <- struct{}{}
 105 }
 106
 107 // retrieveDataReq is the command to retrieve data for a view.
 108 type retrieveDataReq struct {
 109         now time.Time
 110         v   string
 111         c   chan *retrieveDataResp
 112 }
 113
 114 type retrieveDataResp struct {
 115         rows []*Row
 116         err  error
 117 }
 118
 119 func (cmd *retrieveDataReq) handleCommand(w *worker) {
 120         vi, ok := w.views[cmd.v]
 121         if !ok {
 122                 cmd.c <- &retrieveDataResp{
 123                         nil,
 124                         fmt.Errorf("cannot retrieve data; view %q is not registered", cmd.v),
 125                 }
 126                 return
 127         }
 128
 129         if !vi.isSubscribed() {
 130                 cmd.c <- &retrieveDataResp{
 131                         nil,
 132                         fmt.Errorf("cannot retrieve data; view %q has no subscriptions or collection is not forcibly started", cmd.v),
 133                 }
 134                 return
 135         }
 136         cmd.c <- &retrieveDataResp{
 137                 vi.collectedRows(),
 138                 nil,
 139         }
 140 }
 141
 142 // recordReq is the command to record data related to multiple measures
 143 // at once.
 144 type recordReq struct {
 145         tm          *tag.Map
 146         ms          []stats.Measurement
 147         attachments map[string]string
 148         t           time.Time
 149 }
 150
 151 func (cmd *recordReq) handleCommand(w *worker) {
 152         for _, m := range cmd.ms {
 153                 if (m == stats.Measurement{}) { // not registered
 154                         continue
 155                 }
 156                 ref := w.getMeasureRef(m.Measure().Name())
 157                 for v := range ref.views {
 158                         e := &exemplar.Exemplar{
 159                                 Value:       m.Value(),
 160                                 Timestamp:   cmd.t,
 161                                 Attachments: cmd.attachments,
 162                         }
 163                         v.addSample(cmd.tm, e)
 164                 }
 165         }
 166 }
 167
 168 // setReportingPeriodReq is the command to modify the duration between
 169 // reporting the collected data to the registered clients.
 170 type setReportingPeriodReq struct {
 171         d time.Duration
 172         c chan bool
 173 }
 174
 175 func (cmd *setReportingPeriodReq) handleCommand(w *worker) {
 176         w.timer.Stop()
 177         if cmd.d <= 0 {
 178                 w.timer = time.NewTicker(defaultReportingDuration)
 179         } else {
 180                 w.timer = time.NewTicker(cmd.d)
 181         }
 182         cmd.c <- true
 183 }