-
Notifications
You must be signed in to change notification settings - Fork 159
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
pillar: implement NTP sources info messages to the controller
This implements NTP sources eve-api/ptoto/info/ntpsources.proto info messages, which periodically (at least once per 10 min) are sent to the controller. NTP sources data is fetched from the chronyd running on EVE over the unix domain socket with the help of vendor API from the github.com/facebook/time/ntp/chrony package. Currently this API supports only monitoring (RO), not control packages. In the future this can be extended, so `nim` can have full control over the chronyd and update its servers and pools. Signed-off-by: Roman Penyaev <r.peniaev@gmail.com>
- Loading branch information
Showing
8 changed files
with
335 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,303 @@ | ||
| // Copyright (c) 2024 Zededa, Inc. | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package zedagent | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "fmt" | ||
| "net" | ||
| "os" | ||
| "path" | ||
| "time" | ||
|
|
||
| "github.com/golang/protobuf/ptypes" | ||
|
|
||
| chrony "github.com/facebook/time/ntp/chrony" | ||
| "github.com/lf-edge/eve-api/go/info" | ||
| "github.com/lf-edge/eve/pkg/pillar/flextimer" | ||
| "github.com/lf-edge/eve/pkg/pillar/types" | ||
| "google.golang.org/protobuf/proto" | ||
| timestamppb "google.golang.org/protobuf/types/known/timestamppb" | ||
| ) | ||
|
|
||
| const ( | ||
| // Path to unix chrony socket | ||
| unixChronydPath = "/run/chrony/chronyd.sock" | ||
| ) | ||
|
|
||
| // Run a periodic post of the NTP sources information. | ||
| func ntpSourcesTimerTask(ctx *zedagentContext, handleChannel chan interface{}, | ||
| triggerNTPSourcesInfo chan destinationBitset) { | ||
|
|
||
| // Ticker for periodic publishing NTP sources to the controller. | ||
| globalInterval := ctx.globalConfig.GlobalValueInt(types.NTPSourcesInterval) | ||
| interval := time.Duration(globalInterval) * time.Second | ||
| max := float64(interval) | ||
| min := max * 0.3 | ||
| ticker := flextimer.NewRangeTicker(time.Duration(min), time.Duration(max)) | ||
|
|
||
| // Return handles to the caller. | ||
| handleChannel <- ticker | ||
|
|
||
| wdName := agentName + "-ntp" | ||
|
|
||
| // Run a periodic timer so we always update StillRunning | ||
| stillRunning := time.NewTicker(25 * time.Second) | ||
| ctx.ps.StillRunning(wdName, warningTime, errorTime) | ||
| ctx.ps.RegisterFileWatchdog(wdName) | ||
|
|
||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| publishNTPSources(ctx, wdName, ControllerDest) | ||
| case dest := <-triggerNTPSourcesInfo: | ||
| publishNTPSources(ctx, wdName, dest) | ||
| case <-stillRunning.C: | ||
| } | ||
| ctx.ps.StillRunning(wdName, warningTime, errorTime) | ||
| } | ||
| } | ||
|
|
||
| // Called when globalConfig changes. | ||
| // Assumes that the caller has verifier that the interval has changed. | ||
| func updateNTPSourcesTimer(ctx *getconfigContext, globalInterval uint32) { | ||
| if ctx.ntpSourcesTickerHandle == nil { | ||
| log.Warnf("updateNTPSourcesTimer: ticker is still nil") | ||
| return | ||
| } | ||
| interval := time.Duration(globalInterval) * time.Second | ||
| log.Functionf("updateNTPSourcesTimer: interval change to %v", interval) | ||
| max := float64(interval) | ||
| min := max * 0.3 | ||
| flextimer.UpdateRangeTicker(ctx.ntpSourcesTickerHandle, | ||
| time.Duration(min), time.Duration(max)) | ||
| // Force an immediate timeout since timer could have decreased. | ||
| flextimer.TickNow(ctx.ntpSourcesTickerHandle) | ||
| } | ||
|
|
||
| func publishNTPSources(ctx *zedagentContext, wdName string, | ||
| dest destinationBitset) { | ||
| info := getNTPSourcesInfo(ctx) | ||
| if info == nil { | ||
| // Not available. | ||
| return | ||
| } | ||
| start := time.Now() | ||
| publishNTPSourcesToDest(ctx, info, dest) | ||
| ctx.ps.CheckMaxTimeTopic(wdName, "publishNTPSources", start, | ||
| warningTime, errorTime) | ||
| } | ||
|
|
||
| func publishNTPSourcesToDest(ctx *zedagentContext, | ||
| infoNTPSources *info.ZInfoNTPSources, dest destinationBitset) { | ||
|
|
||
| // TODO: we don't support LPS | ||
| if (dest &^ LPSDest) == 0 { | ||
| return | ||
| } | ||
| infoMsg := &info.ZInfoMsg{ | ||
| Ztype: info.ZInfoTypes_ZiNTPSources, | ||
| DevId: devUUID.String(), | ||
| InfoContent: &info.ZInfoMsg_NtpSources{ | ||
| NtpSources: infoNTPSources, | ||
| }, | ||
| AtTimeStamp: ptypes.TimestampNow(), | ||
| } | ||
|
|
||
| log.Functionf("publishNTPSourcesToDest: sending %v", infoMsg) | ||
| data, err := proto.Marshal(infoMsg) | ||
| if err != nil { | ||
| log.Fatal("publishNTPSourcesToController: proto marshaling error: ", err) | ||
| } | ||
| buf := bytes.NewBuffer(data) | ||
| if buf == nil { | ||
| log.Fatal("malloc error") | ||
| } | ||
| size := int64(proto.Size(infoMsg)) | ||
|
|
||
| const bailOnHTTPErr = false | ||
| const withNetTrace = false | ||
| key := "ntpsources:" + devUUID.String() | ||
|
|
||
| // Even for the controller destination we can't stall the queue on error, | ||
| // because this is recurring call, so set @forcePeriodic to true | ||
| forcePeriodic := true | ||
| queueInfoToDest(ctx, dest, key, buf, size, bailOnHTTPErr, withNetTrace, | ||
| forcePeriodic, info.ZInfoTypes_ZiNTPSources) | ||
| } | ||
|
|
||
| // createNTPSource() returns `info.NTPSource`. The code is based on the | ||
| // https://github.com/facebook/time/blob/main/cmd/ntpcheck/checker | ||
| func createNTPSource(s *chrony.ReplySourceData, | ||
| p *chrony.ReplyNTPData, | ||
| n *chrony.ReplyNTPSourceName) (*info.NTPSource, error) { | ||
|
|
||
| // Clear auth and interleaved flag | ||
| flash := s.Flags & chrony.NTPFlagsTests | ||
| // Don't report all flashers if peer is unreachable | ||
| if flash > 0 { | ||
| flash ^= chrony.NTPFlagsTests | ||
| } | ||
| ntpSource := info.NTPSource{ | ||
| Authenticated: (s.Flags & chrony.NTPFlagAuthenticated) != 0, | ||
| Reachable: s.Reachability == 255, // all 8 attempts | ||
| Reachability: uint32(s.Reachability), | ||
| // We have to advance on 1 due to the UNSPECIFIED enum in protobuf | ||
| Mode: info.NTPSourceMode(s.Mode + 1), | ||
| // We have to advance on 1 due to the UNSPECIFIED enum in protobuf | ||
| State: info.NTPSourceState(s.State + 1), | ||
| Flags: uint32(flash), | ||
| // sourceData offset and NTPData offset sign has opposite meaning | ||
| Offset: -1 * s.OrigLatestMeas, | ||
| Poll: int32(s.Poll), | ||
| Stratum: uint32(s.Stratum), | ||
| // Address of the NTP peer, so destination | ||
| DstAddr: s.IPAddr.String(), | ||
| } | ||
| if ntpSource.Mode > info.NTPSourceMode_NTP_SOURCE_MODE_REF { | ||
| ntpSource.Mode = info.NTPSourceMode_NTP_SOURCE_MODE_UNSPECIFIED | ||
| } | ||
| if ntpSource.State > info.NTPSourceState_NTP_SOURCE_STATE_OUTLIER { | ||
| ntpSource.State = info.NTPSourceState_NTP_SOURCE_STATE_UNSPECIFIED | ||
| } | ||
|
|
||
| // Populate data from NTPData struct | ||
| if p != nil { | ||
| refID := chrony.RefidAsHEX(p.RefID) | ||
| // Only stratum 1 servers can have GPS or something else as string refID | ||
| if p.Stratum == 1 { | ||
| refIDStr := chrony.RefidToString(p.RefID) | ||
| if len(refIDStr) > 0 { | ||
| refID = refIDStr | ||
| } | ||
| } | ||
| ntpSource.Leap = uint32(p.Leap) | ||
| ntpSource.Poll = int32(p.Poll) | ||
| // Local address the connection to NTP peer, so source | ||
| ntpSource.SrcAddr = p.LocalAddr.String() | ||
| ntpSource.RefTime = timestamppb.New(p.RefTime) | ||
| ntpSource.Offset = p.Offset | ||
| ntpSource.Dispersion = p.PeerDispersion | ||
| // Missing that info | ||
| ntpSource.SrcPort = 0 | ||
| ntpSource.DstPort = uint32(p.RemotePort) | ||
| ntpSource.RefId = refID | ||
| ntpSource.Jitter = p.PeerDispersion | ||
| ntpSource.RootDelay = p.RootDelay | ||
| ntpSource.Precision = uint32(p.Precision) | ||
| ntpSource.Delay = p.PeerDelay | ||
| ntpSource.RootDisp = p.RootDispersion | ||
| } | ||
| if n != nil { | ||
| // This field is zero padded in chrony, so we need to trim it | ||
| ntpSource.Hostname = string(bytes.TrimRight(n.Name[:], "\x00")) | ||
| } | ||
|
|
||
| return &ntpSource, nil | ||
| } | ||
|
|
||
| type chronyConn struct { | ||
| net.Conn | ||
| local string | ||
| } | ||
|
|
||
| // dialUnixWithChronyd() established connection. The code is a based on the | ||
| // https://github.com/facebook/time/blob/main/cmd/ntpcheck/checker | ||
| func dialUnixWithChronyd(address string) (*chronyConn, error) { | ||
| base, _ := path.Split(address) | ||
| local := path.Join(base, fmt.Sprintf("chronyc.%d.sock", os.Getpid())) | ||
| conn, err := net.DialUnix("unixgram", | ||
| &net.UnixAddr{Name: local, Net: "unixgram"}, | ||
| &net.UnixAddr{Name: address, Net: "unixgram"}, | ||
| ) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if err := os.Chmod(local, 0600); err != nil { | ||
| conn.Close() | ||
| return nil, err | ||
| } | ||
| return &chronyConn{Conn: conn, local: local}, nil | ||
| } | ||
|
|
||
| // getNTPSourcesInfo() returns `info.ZInfoNTPSources`. The code is based on the | ||
| // https://github.com/facebook/time/blob/main/cmd/ntpcheck/checker | ||
| func getNTPSourcesInfo(ctx *zedagentContext) *info.ZInfoNTPSources { | ||
| conn, err := dialUnixWithChronyd(unixChronydPath) | ||
| if err != nil { | ||
| log.Errorf("getNTPSourcesInfo: can't connect to chronyd: %v", err) | ||
| return nil | ||
| } | ||
| defer func() { | ||
| conn.Close() | ||
| os.Remove(conn.local) | ||
| }() | ||
|
|
||
| client := chrony.Client{Sequence: 1, Connection: conn} | ||
| sourcesReq := chrony.NewSourcesPacket() | ||
| packet, err := client.Communicate(sourcesReq) | ||
| if err != nil { | ||
| log.Errorf("getNTPSourcesInfo: failed to get 'sources' response: %v", err) | ||
| return nil | ||
| } | ||
| sources, ok := packet.(*chrony.ReplySources) | ||
| if !ok { | ||
| log.Errorf("getNTPSourcesInfo: failed to convert to reply: %v", err) | ||
| return nil | ||
| } | ||
|
|
||
| info := info.ZInfoNTPSources{} | ||
|
|
||
| for i := 0; i < sources.NSources; i++ { | ||
| sourceDataReq := chrony.NewSourceDataPacket(int32(i)) | ||
| packet, err = client.Communicate(sourceDataReq) | ||
| if err != nil { | ||
| log.Errorf("getNTPSourcesInfo: failed to get 'sourcedata' response for source #%d, err %v", i, err) | ||
| return nil | ||
| } | ||
| sourceData, ok := packet.(*chrony.ReplySourceData) | ||
| if !ok { | ||
| log.Errorf("getNTPSourcesInfo: got wrong 'sourcedata' response %+v", packet) | ||
| return nil | ||
| } | ||
|
|
||
| // get ntpdata when using a unix socket | ||
| var ntpData *chrony.ReplyNTPData | ||
| if sourceData.Mode != chrony.SourceModeRef { | ||
| ntpDataReq := chrony.NewNTPDataPacket(sourceData.IPAddr) | ||
| packet, err = client.Communicate(ntpDataReq) | ||
| if err != nil { | ||
| log.Errorf("getNTPSourcesInfo: failed to get 'ntpdata' response for source #%d", i) | ||
| return nil | ||
| } | ||
| ntpData, ok = packet.(*chrony.ReplyNTPData) | ||
| if !ok { | ||
| log.Errorf("getNTPSourcesInfo: got wrong 'ntpdata' response %+v", packet) | ||
| return nil | ||
| } | ||
| } | ||
| var ntpSourceName *chrony.ReplyNTPSourceName | ||
| if sourceData.Mode != chrony.SourceModeRef { | ||
| ntpSourceNameReq := chrony.NewNTPSourceNamePacket(sourceData.IPAddr) | ||
| packet, err = client.Communicate(ntpSourceNameReq) | ||
| if err != nil { | ||
| log.Errorf("getNTPSourcesInfo: failed to get 'sourcename' response for source #%d", i) | ||
| return nil | ||
| } | ||
| ntpSourceName, ok = packet.(*chrony.ReplyNTPSourceName) | ||
| if !ok { | ||
| log.Errorf("getNTPSourcesInfo: got wrong 'sourcename' response %+v", packet) | ||
| return nil | ||
| } | ||
| } | ||
| ntpSource, err := createNTPSource(sourceData, ntpData, ntpSourceName) | ||
| if err != nil { | ||
| log.Errorf("getNTPSourcesInfo: failed to create Peer structure from response packet for peer=%s", sourceData.IPAddr) | ||
| return nil | ||
| } | ||
| info.Sources = append(info.Sources, ntpSource) | ||
| } | ||
|
|
||
| return &info | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.